This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new a731c705a9 Action IO events are linked to the workflow and not the
action, fixes #7522 (#7523)
a731c705a9 is described below
commit a731c705a9ad4eee9a093a0bd2e2a8b426baa579
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Tue Jul 14 13:15:52 2026 +0200
Action IO events are linked to the workflow and not the action, fixes
#7522 (#7523)
---
.../apache/hop/lineage/LineageFileIoEmitter.java | 23 ++---------
.../apache/hop/lineage/LineageHttpIoEmitter.java | 25 ++----------
.../hop/lineage/LineageRunLifecycleEmitter.java | 43 +++++++++++++++++++++
.../apache/hop/lineage/context/LineageContext.java | 7 +++-
.../hop/lineage/LineageFileIoEmitterTest.java | 44 ++++++++++++++++++++++
.../hop/lineage/LineageHttpIoEmitterTest.java | 8 ++++
6 files changed, 107 insertions(+), 43 deletions(-)
diff --git
a/engine/src/main/java/org/apache/hop/lineage/LineageFileIoEmitter.java
b/engine/src/main/java/org/apache/hop/lineage/LineageFileIoEmitter.java
index 5fd201fdd9..9ac829a081 100644
--- a/engine/src/main/java/org/apache/hop/lineage/LineageFileIoEmitter.java
+++ b/engine/src/main/java/org/apache/hop/lineage/LineageFileIoEmitter.java
@@ -20,8 +20,6 @@ package org.apache.hop.lineage;
import org.apache.commons.vfs2.FileObject;
import org.apache.hop.core.util.Utils;
import org.apache.hop.lineage.context.LineageContext;
-import org.apache.hop.lineage.context.LineagePortableFilename;
-import org.apache.hop.lineage.context.LineageSubjectType;
import org.apache.hop.lineage.hub.LineageHub;
import org.apache.hop.lineage.model.FileIoContentSchema;
import org.apache.hop.lineage.model.FileIoLineagePayload;
@@ -132,25 +130,10 @@ public final class LineageFileIoEmitter {
private static void emit(
IWorkflowEngine<WorkflowMeta> workflow, IAction action,
FileIoLineagePayload payload) {
- WorkflowMeta meta = workflow.getWorkflowMeta();
- String workflowName = meta != null ? meta.getName() : null;
- String filename = meta != null ? meta.getFilename() : null;
-
- LineageContext.Builder ctx =
- LineageContext.builder()
- .subjectType(LineageSubjectType.ACTION)
- .logChannelId(workflow.getLogChannelId())
- .workflowName(workflowName)
- .actionName(action.getName());
- if (!Utils.isEmpty(filename)) {
- ctx.hopFilename(filename);
- ctx.hopFilenamePortableKey(LineagePortableFilename.portableKey(filename,
workflow));
- }
- ctx.putAttribute("workflowLogChannelId", workflow.getLogChannelId());
- if (!Utils.isEmpty(action.getPluginId())) {
- ctx.putAttribute("actionPluginId", action.getPluginId());
+ LineageContext.Builder ctx =
LineageRunLifecycleEmitter.actionContextBuilder(workflow, action);
+ if (ctx == null) {
+ return;
}
-
LineageHub.getInstance().emit(LineageEvent.of(LineageEventKind.FILE_IO,
ctx.build(), payload));
}
diff --git
a/engine/src/main/java/org/apache/hop/lineage/LineageHttpIoEmitter.java
b/engine/src/main/java/org/apache/hop/lineage/LineageHttpIoEmitter.java
index 73b52eb595..62c75f0991 100644
--- a/engine/src/main/java/org/apache/hop/lineage/LineageHttpIoEmitter.java
+++ b/engine/src/main/java/org/apache/hop/lineage/LineageHttpIoEmitter.java
@@ -19,8 +19,6 @@ package org.apache.hop.lineage;
import org.apache.hop.core.util.Utils;
import org.apache.hop.lineage.context.LineageContext;
-import org.apache.hop.lineage.context.LineagePortableFilename;
-import org.apache.hop.lineage.context.LineageSubjectType;
import org.apache.hop.lineage.hub.LineageHub;
import org.apache.hop.lineage.model.HttpLineagePayload;
import org.apache.hop.lineage.model.LineageEvent;
@@ -54,28 +52,13 @@ public final class LineageHttpIoEmitter {
public static void emitWorkflowActionHttpIo(
IWorkflowEngine<WorkflowMeta> workflow, IAction action,
HttpLineagePayload payload) {
- if (workflow == null || action == null || payload == null) {
+ if (payload == null) {
return;
}
- WorkflowMeta meta = workflow.getWorkflowMeta();
- String workflowName = meta != null ? meta.getName() : null;
- String filename = meta != null ? meta.getFilename() : null;
-
- LineageContext.Builder ctx =
- LineageContext.builder()
- .subjectType(LineageSubjectType.ACTION)
- .logChannelId(workflow.getLogChannelId())
- .workflowName(workflowName)
- .actionName(action.getName());
- if (!Utils.isEmpty(filename)) {
- ctx.hopFilename(filename);
- ctx.hopFilenamePortableKey(LineagePortableFilename.portableKey(filename,
workflow));
- }
- ctx.putAttribute("workflowLogChannelId", workflow.getLogChannelId());
- if (!Utils.isEmpty(action.getPluginId())) {
- ctx.putAttribute("actionPluginId", action.getPluginId());
+ LineageContext.Builder ctx =
LineageRunLifecycleEmitter.actionContextBuilder(workflow, action);
+ if (ctx == null) {
+ return;
}
-
LineageHub.getInstance().emit(LineageEvent.of(LineageEventKind.HTTP_IO,
ctx.build(), payload));
}
}
diff --git
a/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java
b/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java
index 148a1a0383..40a700aecd 100644
---
a/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java
+++
b/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java
@@ -19,6 +19,7 @@ package org.apache.hop.lineage;
import org.apache.hop.core.Const;
import org.apache.hop.core.Result;
+import org.apache.hop.core.logging.ILogChannel;
import org.apache.hop.core.util.Utils;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.lineage.context.LineageContext;
@@ -36,6 +37,7 @@ import
org.apache.hop.pipeline.transform.TransformMetaDataCombi;
import org.apache.hop.workflow.WorkflowExecutionExtension;
import org.apache.hop.workflow.WorkflowMeta;
import org.apache.hop.workflow.action.ActionMeta;
+import org.apache.hop.workflow.action.IAction;
import org.apache.hop.workflow.engine.IWorkflowEngine;
/**
@@ -90,6 +92,47 @@ public final class LineageRunLifecycleEmitter {
return ctx;
}
+ /**
+ * Correlation context for a workflow action instance that is currently
executing — used for the
+ * file and HTTP I/O events emitted from inside {@link IAction#execute}.
+ *
+ * <p>The log channel id is the action instance's own channel, not the
workflow's: the workflow
+ * clones the action and hands the clone a fresh log channel per execution,
so this is the id that
+ * correlates an I/O event with the action's own lifecycle events and log
lines. The workflow's
+ * channel remains available as the {@code workflowLogChannelId} attribute.
Falls back to the
+ * workflow's channel only when the action has no channel of its own.
+ */
+ public static LineageContext.Builder actionContextBuilder(
+ IWorkflowEngine<WorkflowMeta> workflow, IAction action) {
+ if (workflow == null || action == null) {
+ return null;
+ }
+ WorkflowMeta meta = workflow.getWorkflowMeta();
+ String workflowName = meta != null ? meta.getName() : null;
+ String filename = meta != null ? meta.getFilename() : null;
+
+ ILogChannel actionLog = action.getLogChannel();
+ String actionLogChannelId = actionLog != null ?
actionLog.getLogChannelId() : null;
+
+ LineageContext.Builder ctx =
+ LineageContext.builder()
+ .subjectType(LineageSubjectType.ACTION)
+ .logChannelId(
+ !Utils.isEmpty(actionLogChannelId)
+ ? actionLogChannelId
+ : workflow.getLogChannelId())
+ .workflowName(workflowName)
+ .actionName(action.getName());
+ if (!Utils.isEmpty(filename)) {
+ fillFilenameFields(ctx, filename, workflow);
+ }
+ ctx.putAttribute("workflowLogChannelId", workflow.getLogChannelId());
+ if (!Utils.isEmpty(action.getPluginId())) {
+ ctx.putAttribute("actionPluginId", action.getPluginId());
+ }
+ return ctx;
+ }
+
static LineageContext.Builder transformContextBuilder(TransformMetaDataCombi
combi) {
if (combi == null || combi.transform == null) {
return null;
diff --git
a/engine/src/main/java/org/apache/hop/lineage/context/LineageContext.java
b/engine/src/main/java/org/apache/hop/lineage/context/LineageContext.java
index 472b1ffb38..0d9e8034a1 100644
--- a/engine/src/main/java/org/apache/hop/lineage/context/LineageContext.java
+++ b/engine/src/main/java/org/apache/hop/lineage/context/LineageContext.java
@@ -38,8 +38,11 @@ import lombok.Getter;
* channel)
* <li><b>TRANSFORM</b> — {@code pipelineName}, {@code transformName},
{@code copyNr}, and {@code
* logChannelId} (the transform's log channel)
- * <li><b>ACTION</b> — {@code workflowName}, {@code actionName}, and {@code
logChannelId} (varies
- * by lifecycle phase between the workflow's and the action instance's
log channel)
+ * <li><b>ACTION</b> — {@code workflowName}, {@code actionName}, and {@code
logChannelId} (the
+ * executing action instance's log channel; the workflow's channel is
always available as the
+ * {@code workflowLogChannelId} attribute). The one exception is the
{@code STARTED} run
+ * lifecycle phase: the action instance does not exist yet at that
point, so {@code
+ * logChannelId} carries the workflow's channel.
* <li><b>OTHER</b> — sentinel for events that do not map to a single subject
* </ul>
*
diff --git
a/engine/src/test/java/org/apache/hop/lineage/LineageFileIoEmitterTest.java
b/engine/src/test/java/org/apache/hop/lineage/LineageFileIoEmitterTest.java
index 34cd62ee63..4d73f083c3 100644
--- a/engine/src/test/java/org/apache/hop/lineage/LineageFileIoEmitterTest.java
+++ b/engine/src/test/java/org/apache/hop/lineage/LineageFileIoEmitterTest.java
@@ -60,9 +60,13 @@ class LineageFileIoEmitterTest {
when(wf.getWorkflowMeta()).thenReturn(meta);
when(wf.getLogChannelId()).thenReturn("wf-log-1");
+ ILogChannel actionLog = mock(ILogChannel.class);
+ when(actionLog.getLogChannelId()).thenReturn("action-log-1");
+
ActionBase action = mock(ActionBase.class);
when(action.getName()).thenReturn("Move stuff");
when(action.getPluginId()).thenReturn("MOVE_FILES");
+ when(action.getLogChannel()).thenReturn(actionLog);
try (MockedStatic<LineageHub> staticHub =
Mockito.mockStatic(LineageHub.class)) {
staticHub.when(LineageHub::getInstance).thenReturn(hub);
@@ -94,6 +98,15 @@ class LineageFileIoEmitterTest {
if (!"Move stuff".equals(e.getContext().getActionName())) {
return false;
}
+ // The subject is the action, so the log channel must be the
action instance's,
+ // not the workflow's.
+ if
(!"action-log-1".equals(e.getContext().getLogChannelId())) {
+ return false;
+ }
+ if (!"wf-log-1"
+
.equals(e.getContext().getAttributes().get("workflowLogChannelId"))) {
+ return false;
+ }
if (!(e.getPayload() instanceof FileIoLineagePayload p)) {
return false;
}
@@ -106,6 +119,37 @@ class LineageFileIoEmitterTest {
}));
}
+ @Test
+ void
emitWorkflowActionFileIo_fallsBackToWorkflowLogChannelWhenActionHasNone() {
+ LineageHub hub = mock(LineageHub.class);
+
+ WorkflowMeta meta = new WorkflowMeta();
+ meta.setName("wf1");
+
+ @SuppressWarnings("unchecked")
+ IWorkflowEngine<WorkflowMeta> wf = mock(IWorkflowEngine.class);
+ when(wf.getWorkflowMeta()).thenReturn(meta);
+ when(wf.getLogChannelId()).thenReturn("wf-log-1");
+
+ ActionBase action = mock(ActionBase.class);
+ when(action.getName()).thenReturn("Move stuff");
+ when(action.getLogChannel()).thenReturn(null);
+
+ try (MockedStatic<LineageHub> staticHub =
Mockito.mockStatic(LineageHub.class)) {
+ staticHub.when(LineageHub::getInstance).thenReturn(hub);
+
+ LineageFileIoEmitter.emitWorkflowActionFileIo(
+ wf, action, FileIoOperation.DELETE, "file:///data/in.csv", null,
null, true, null);
+ }
+
+ verify(hub)
+ .emit(
+ argThat(
+ (LineageEvent e) ->
+ e.getContext().getSubjectType() ==
LineageSubjectType.ACTION
+ &&
"wf-log-1".equals(e.getContext().getLogChannelId())));
+ }
+
@Test
void emitWorkflowActionFileIo_withFileObjects_setsUriFromFileName() throws
Exception {
LineageHub hub = mock(LineageHub.class);
diff --git
a/engine/src/test/java/org/apache/hop/lineage/LineageHttpIoEmitterTest.java
b/engine/src/test/java/org/apache/hop/lineage/LineageHttpIoEmitterTest.java
index 0cda46e412..71680554e8 100644
--- a/engine/src/test/java/org/apache/hop/lineage/LineageHttpIoEmitterTest.java
+++ b/engine/src/test/java/org/apache/hop/lineage/LineageHttpIoEmitterTest.java
@@ -98,9 +98,13 @@ class LineageHttpIoEmitterTest {
when(wf.getWorkflowMeta()).thenReturn(meta);
when(wf.getLogChannelId()).thenReturn("wf-log");
+ ILogChannel actionLog = mock(ILogChannel.class);
+ when(actionLog.getLogChannelId()).thenReturn("action-log");
+
ActionBase action = mock(ActionBase.class);
when(action.getName()).thenReturn("HTTP");
when(action.getPluginId()).thenReturn("HTTP");
+ when(action.getLogChannel()).thenReturn(actionLog);
HttpLineagePayload payload =
new HttpLineagePayload(
@@ -117,6 +121,10 @@ class LineageHttpIoEmitterTest {
(LineageEvent e) ->
e.getKind() == LineageEventKind.HTTP_IO
&& e.getContext().getSubjectType() ==
LineageSubjectType.ACTION
+ // subject is the action, so its own log channel
identifies it
+ &&
"action-log".equals(e.getContext().getLogChannelId())
+ && "wf-log"
+
.equals(e.getContext().getAttributes().get("workflowLogChannelId"))
&& e.getPayload() instanceof HttpLineagePayload p
&& p.getRequestBytes() == 100L));
}
