Skip to content

Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005 #9035

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hansva merged 2 commits into from
Oct 24, 2026
+353 −45
Merged

Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005 #9035

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3d699ef
Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005
bamaer Oct 20, 2026
a04e82e
Count completion failures before releasing waiters, flush lineage on …
bamaer Oct 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 88 additions & 45 deletions engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1376,7 +1376,7 @@ private void startThreadsInternal() throws HopException {
try {
fireExecutionFinishedListeners();
} catch (Exception e) {
transform.setErrors(transform.getErrors() + 1L);
// fireExecutionFinishedListeners() already counted this on the pipeline
log.logError(
getName()
+ " : "
Expand Down Expand Up @@ -1406,7 +1406,11 @@ private void startThreadsInternal() throws HopException {
}
};

// Make sure this is called first!
// Register the completion detection before the other listeners, so the pipeline's own
// bookkeeping (finished flag, end date, error handling) does not wait on them. The
// TransformFinished extension point is inserted ahead of it further down, before the
// transform threads start: a transform has to finish announcing itself before the pipeline
// can conclude that the last one is done.
//
if (sid.transform instanceof BaseTransform baseTransform) {
baseTransform.getTransformFinishedListeners().add(0, finishedListener);
Expand Down Expand Up @@ -1499,29 +1503,19 @@ public void run() {
RunThread runThread = new RunThread(combi);
Thread thread = new Thread(runThread);
thread.setName(getName() + " - " + combi.transformName);
ExtensionPointHandler.callExtensionPoint(
log, this, HopExtensionPoint.TransformBeforeStart.id, combi);
// Call an extension point at the end of the transform
//
combi.transform.addTransformFinishedListener(
(pipeline, transformMeta, transform) -> {
try {
ExtensionPointHandler.callExtensionPoint(
log, this, HopExtensionPoint.TransformFinished.id, combi);
} catch (HopException e) {
throw new HopRuntimeException(
"Unexpected error in calling extension point upon transform finish", e);
}
});

registerTransformExtensionPoints(combi);
thread.start();
}
break;

case SingleThreaded:
// Don't do anything, this needs to be handled by the pipeline
// executor!
// The executor runs these transforms one at a time, so there are no threads to start here,
// but they announce themselves like any other transform. The executor only stops them when
// it disposes them, so their finished announcement lands at the end of the run.
//
for (final TransformMetaDataCombi combi : transforms) {
registerTransformExtensionPoints(combi);
}
break;
default:
break;
Expand All @@ -1546,6 +1540,43 @@ public void run() {
}
}

/**
* Announce a transform that is about to start, and register the extension point that announces it
* when it finishes.
*
* <p>The finished listener is inserted ahead of the completion detection registered above: the
* pipeline fires its completion, and with it whatever that triggers (flushing queued lineage, for
* example), the moment the last transform stops, so a transform has to finish announcing itself
* before that. Called before execution starts, so no transform can beat the registration.
*/
private void registerTransformExtensionPoints(TransformMetaDataCombi combi) throws HopException {
ExtensionPointHandler.callExtensionPoint(
log, this, HopExtensionPoint.TransformBeforeStart.id, combi);

// This listener runs ahead of the completion detection, so it must never throw: a failure here
// would skip the detection for this transform and leave waitUntilFinished() waiting forever.
ITransformFinishedListener transformFinishedExtensionPoint =
(pipeline, transformMeta, transform) -> {
try {
ExtensionPointHandler.callExtensionPoint(
log, this, HopExtensionPoint.TransformFinished.id, combi);
} catch (Exception e) {
log.logError(
"Unexpected error in calling extension point upon transform finish of '"
+ combi.transformName
+ "'",
e);
}
};
if (combi.transform instanceof BaseTransform baseTransform) {
baseTransform.getTransformFinishedListeners().add(0, transformFinishedExtensionPoint);
} else {
// The interface has no insert, so an implementation that does not extend BaseTransform keeps
// the old order and with it a late last event.
combi.transform.addTransformFinishedListener(transformFinishedExtensionPoint);
}
}

/**
* @deprecated Make attempt to fire all registered finished listeners if possible.
* @throws HopException if any errors occur during notification
Expand All @@ -1558,44 +1589,56 @@ public void firePipelineExecutionFinishedListeners() throws HopException {

@Override
public void fireExecutionFinishedListeners() throws HopException {
HopException listenerException = null;
synchronized (executionFinishedListeners) {
if (!executionFinishedListeners.isEmpty()) {
// prevent Exception from one listener to block others execution
List<HopException> badGuys = new ArrayList<>(executionFinishedListeners.size());
for (IExecutionFinishedListener<IPipelineEngine<PipelineMeta>> listener :
executionFinishedListeners) {
try {
listener.finished(this);
} catch (HopException e) {
badGuys.add(e);
try {
HopException listenerException = null;
synchronized (executionFinishedListeners) {
if (!executionFinishedListeners.isEmpty()) {
// prevent Exception from one listener to block others execution
List<HopException> badGuys = new ArrayList<>(executionFinishedListeners.size());
for (IExecutionFinishedListener<IPipelineEngine<PipelineMeta>> listener :
executionFinishedListeners) {
try {
listener.finished(this);
} catch (HopException e) {
badGuys.add(e);
}
}
if (!badGuys.isEmpty()) {
// FIFO
listenerException = badGuys.get(0);
}
}
if (!badGuys.isEmpty()) {
// FIFO
listenerException = badGuys.get(0);
}
}
}

try {
// Now the status and everything else is set correctly. We've completed the pipeline.
//
pipelineCompleted();

// Also call an extension point in case plugins want to play along
// Call the completion extension point before releasing anything blocked in
// waitUntilFinished(): plugins here may still have work to do - flushing lineage, for
// example - while a waiting caller (hop-run, for one) can leave the JVM the moment it is
// released.
//
ExtensionPointHandler.callExtensionPoint(
log, this, HopExtensionPoint.PipelineCompleted.id, this);

if (listenerException != null) {
throw listenerException;
}
} catch (Exception e) {
// Count the failure before the waiters are released: a released caller reads the result
// straight away, hop-run its exit code for one.
//
errors.incrementAndGet();
throw e;
} finally {
// Now the status and everything else is set correctly. We've completed the pipeline.
//
// Always release the waiters, even when a listener or extension point failed: that must never
// be able to deadlock a caller. The failure itself still propagates.
//
pipelineCompleted();

@mattcasters mattcasters Oct 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[suggestion] Releasing the waiter here happens before a PipelineCompleted failure is visible on the result. The exception leaves fireExecutionFinishedListeners() only after this finally, including releaseVfsNamespace(), and the transform-finished listener increments the transform error count only after that. HopRun reads getResult().getNrErrors() as soon as waitUntilFinished() returns, so a throwing extension point can still produce exit code 0. ExtensionPointMap does not wrap runtime exceptions, so those take the same path.

Suggestion: Record the failure on the pipeline errors count before pipelineCompleted(). getResult() already adds that field. Drop the extra setErrors in the transform listener for this same failure, or the run is counted twice. Catch Exception, not only HopException.


// Only now: everything above can still touch files of this namespace, and closing it
// invalidates every file object resolved through it - the result files carry those.
releaseVfsNamespace();
}

if (listenerException != null) {
throw listenerException;
}
}

public void pipelineCompleted() throws HopException {
Expand Down
22 changes: 22 additions & 0 deletions engine/src/main/java/org/apache/hop/run/HopRun.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import org.apache.hop.core.logging.HopLogStore;
import org.apache.hop.core.plugins.JarCache;
import org.apache.hop.hop.Hop;
import org.apache.hop.lineage.hub.LineageHub;
import org.apache.hop.metadata.api.IHasHopMetadataProvider;
import org.apache.hop.metadata.util.HopMetadataInstance;
import org.apache.hop.metadata.util.HopMetadataUtil;
Expand Down Expand Up @@ -92,6 +93,9 @@ public static void main(String[] args) {
System.exit(1);
} else {
hopRun.run();

shutdownLineage();

System.out.println("HopRun exit.");
if (hopRun.isFinishedWithoutError()) {
System.exit(0);
Expand All @@ -102,20 +106,38 @@ public static void main(String[] args) {
} catch (ParameterException e) {
System.err.println(e.getMessage());
hopRun.cmd.usage(System.err);
shutdownLineage();
System.exit(9);
} catch (ExecutionException e) {
System.err.println("Error found during execution!");
System.err.println(Const.getStackTracker(e));

shutdownLineage();
System.exit(1);
} catch (Exception e) {
System.err.println("General error found, something went horribly wrong!");
System.err.println(Const.getStackTracker(e));

shutdownLineage();
System.exit(2);
}
}

/**
* Deliver whatever lineage is still queued before the JVM goes away: the hub hands its queue to
* the sinks and the sinks drain theirs (bounded by their drain timeout). Without it a short run
* loses its terminal events, which the sinks send last, and a failed run loses the events emitted
* before it failed. Call it before every System.exit() once a run may have started. Safe to call
* when lineage is disabled, and a failure here never changes the exit code.
*/
private static void shutdownLineage() {
try {
LineageHub.getInstance().shutdown();
} catch (Exception e) {
System.err.println("Error delivering the remaining lineage events: " + e.getMessage());
}
}

public void applySystemProperties() {
// Set some System properties if there were any
//
Expand Down
Loading
Loading