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 45d652eaf6 Run PipelineCompleted before releasing waitUntilFinished(),
fixes #9005 (#9035)
45d652eaf6 is described below
commit 45d652eaf63451028feedd9617a17dd312c4e113
Author: Bart Maertens <[email protected]>
AuthorDate: Sat Oct 24 09:40:50 2026 +0200
Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005
(#9035)
* Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005
- call the PipelineCompleted extension point before releasing waiters, and
release them in a finally
- register the TransformFinished extension point ahead of the pipeline's
completion detection
- fire TransformBeforeStart/TransformFinished for the SingleThreaded
pipeline type
- flush the lineage hub in HopRun before System.exit
* Count completion failures before releasing waiters, flush lineage on
hop-run error exits, #9005
---
.../java/org/apache/hop/pipeline/Pipeline.java | 133 +++++++----
.../src/main/java/org/apache/hop/run/HopRun.java | 22 ++
.../java/org/apache/hop/pipeline/PipelineTest.java | 243 +++++++++++++++++++++
3 files changed, 353 insertions(+), 45 deletions(-)
diff --git a/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
b/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
index ee4a3de3e7..737fd560b2 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
@@ -1376,7 +1376,7 @@ public abstract class Pipeline
try {
fireExecutionFinishedListeners();
} catch (Exception e) {
- transform.setErrors(transform.getErrors() + 1L);
+ // fireExecutionFinishedListeners() already counted this on
the pipeline
log.logError(
getName()
+ " : "
@@ -1406,7 +1406,11 @@ public abstract class Pipeline
}
};
- // 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);
@@ -1499,29 +1503,19 @@ public abstract class Pipeline
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;
@@ -1546,6 +1540,43 @@ public abstract class Pipeline
}
}
+ /**
+ * 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
@@ -1558,44 +1589,56 @@ public abstract class Pipeline
@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();
+
// 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 {
diff --git a/engine/src/main/java/org/apache/hop/run/HopRun.java
b/engine/src/main/java/org/apache/hop/run/HopRun.java
index e0655d6a20..a45003a9f3 100644
--- a/engine/src/main/java/org/apache/hop/run/HopRun.java
+++ b/engine/src/main/java/org/apache/hop/run/HopRun.java
@@ -27,6 +27,7 @@ import org.apache.hop.core.config.plugin.ConfigPlugin;
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;
@@ -92,6 +93,9 @@ public class HopRun extends HopRunBase implements Runnable,
IHasHopMetadataProvi
System.exit(1);
} else {
hopRun.run();
+
+ shutdownLineage();
+
System.out.println("HopRun exit.");
if (hopRun.isFinishedWithoutError()) {
System.exit(0);
@@ -102,20 +106,38 @@ public class HopRun extends HopRunBase implements
Runnable, IHasHopMetadataProvi
} 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
//
diff --git a/engine/src/test/java/org/apache/hop/pipeline/PipelineTest.java
b/engine/src/test/java/org/apache/hop/pipeline/PipelineTest.java
index 4027a1fe51..84ac02e274 100644
--- a/engine/src/test/java/org/apache/hop/pipeline/PipelineTest.java
+++ b/engine/src/test/java/org/apache/hop/pipeline/PipelineTest.java
@@ -27,19 +27,33 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hop.core.Const;
import org.apache.hop.core.HopEnvironment;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.exception.HopRuntimeException;
+import org.apache.hop.core.extension.ExtensionPointMap;
+import org.apache.hop.core.extension.HopExtensionPoint;
+import org.apache.hop.core.extension.IExtensionPoint;
import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.logging.LoggingObject;
+import org.apache.hop.core.plugins.IClassLoadingPlugin;
+import org.apache.hop.core.plugins.IPlugin;
+import org.apache.hop.core.variables.Variables;
import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
import org.apache.hop.pipeline.engine.IPipelineEngine;
import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
import org.apache.hop.pipeline.transform.ITransform;
import org.apache.hop.pipeline.transform.ITransformData;
import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transforms.dummy.DummyMeta;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -150,6 +164,229 @@ class PipelineTest {
assertThrows(HopException.class, line::fireExecutionFinishedListeners);
}
+ /**
+ * A caller blocked in {@link Pipeline#waitUntilFinished()} must not be
released before the
+ * PipelineCompleted extension point has run. Extension points there still
have work to flush -
+ * lineage for one - and a released caller (hop-run, for example) can leave
the JVM immediately.
+ */
+ @Test
+ void pipelineCompletedExtensionPointRunsBeforeWaitersAreReleased() throws
Exception {
+ List<String> order = Collections.synchronizedList(new ArrayList<>());
+ CountDownLatch waiterReleased = new CountDownLatch(1);
+
+ TestExtensionPointPlugin plugin =
+ registerExtensionPoint(
+ HopExtensionPoint.PipelineCompleted.id,
+ () -> {
+ order.add("extension-point");
+ // If the waiter was released first (the bug) it records itself
while this runs.
+ // Waiting for that here keeps the assertion deterministic
instead of a race.
+ try {
+ waiterReleased.await(400, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ try {
+ Pipeline line = new LocalPipelineEngine(new PipelineMeta());
+ line.setLogChannel(mock(ILogChannel.class));
+ line.prepareExecution();
+
+ Thread waiter =
+ new Thread(
+ () -> {
+ // waitUntilFinished() returns at once while startThreads()
has not created its
+ // queue yet. The queue exists once the pipeline is flagged as
running, and a late
+ // waiter still finds the signal already on it.
+ while (!line.isRunning() && !line.isFinished()) {
+ Thread.onSpinWait();
+ }
+ line.waitUntilFinished();
+ order.add("waiter");
+ waiterReleased.countDown();
+ });
+ waiter.start();
+
+ // A pipeline without transforms completes in startThreads(), on this
thread.
+ line.startThreads();
+
+ waiter.join(10_000);
+ assertEquals(List.of("extension-point", "waiter"), order);
+ } finally {
+ ExtensionPointMap.getInstance().removeExtensionPoint(plugin);
+ }
+ }
+
+ /**
+ * A transform has to finish announcing itself before the pipeline concludes
that the last
+ * transform is done. A plugin emitting its final events on
TransformFinished - lineage does -
+ * otherwise queues them after the pipeline has already fired its completion
and flushed.
+ */
+ @Test
+ void transformFinishedExtensionPointRunsBeforePipelineCompleted() throws
Exception {
+ List<String> order = Collections.synchronizedList(new ArrayList<>());
+
+ TestExtensionPointPlugin transformFinished =
+ registerExtensionPoint(
+ HopExtensionPoint.TransformFinished.id, () ->
order.add("transform-finished"));
+ TestExtensionPointPlugin pipelineCompleted =
+ registerExtensionPoint(
+ HopExtensionPoint.PipelineCompleted.id, () ->
order.add("pipeline-completed"));
+
+ try {
+ PipelineMeta meta = new PipelineMeta();
+ meta.setName("ordering test");
+ meta.addTransform(new TransformMeta("A", new DummyMeta()));
+
+ LocalPipelineEngine run =
+ new LocalPipelineEngine(meta, new Variables(), new
LoggingObject("test"));
+ run.prepareExecution();
+ run.startThreads();
+ run.waitUntilFinished();
+
+ assertEquals(List.of("transform-finished", "pipeline-completed"), order);
+ } finally {
+ ExtensionPointMap.getInstance().removeExtensionPoint(transformFinished);
+ ExtensionPointMap.getInstance().removeExtensionPoint(pipelineCompleted);
+ }
+ }
+
+ /**
+ * The TransformFinished extension point runs ahead of the pipeline's
completion detection, so a
+ * plugin that fails there must not keep the pipeline from finishing.
+ */
+ @Test
+ void failingTransformFinishedExtensionPointDoesNotHangThePipeline() throws
Exception {
+ TestExtensionPointPlugin transformFinished =
+ registerExtensionPoint(
+ HopExtensionPoint.TransformFinished.id,
+ () -> {
+ throw new IllegalStateException("broken plugin");
+ });
+
+ try {
+ PipelineMeta meta = new PipelineMeta();
+ meta.setName("failing extension point test");
+ meta.addTransform(new TransformMeta("A", new DummyMeta()));
+
+ LocalPipelineEngine run =
+ new LocalPipelineEngine(meta, new Variables(), new
LoggingObject("test"));
+ run.prepareExecution();
+ run.startThreads();
+
+ Thread waiter = new Thread(run::waitUntilFinished);
+ waiter.start();
+ waiter.join(10_000);
+
+ assertFalse(waiter.isAlive(), "waitUntilFinished() did not return");
+ assertTrue(run.isFinished());
+ } finally {
+ ExtensionPointMap.getInstance().removeExtensionPoint(transformFinished);
+ }
+ }
+
+ /**
+ * A failing PipelineCompleted extension point has to show up in the result
by the time the
+ * waiters are released: hop-run reads the error count, and with it its exit
code, straight away.
+ * The failure is counted once.
+ */
+ @Test
+ void
failingPipelineCompletedExtensionPointIsCountedBeforeWaitersAreReleased()
throws Exception {
+ TestExtensionPointPlugin pipelineCompleted =
+ registerExtensionPoint(
+ HopExtensionPoint.PipelineCompleted.id,
+ () -> {
+ throw new IllegalStateException("broken plugin");
+ });
+
+ try {
+ PipelineMeta meta = new PipelineMeta();
+ meta.setName("failing completion test");
+ meta.addTransform(new TransformMeta("A", new DummyMeta()));
+
+ // Look at the result the moment the waiters are released: a waiter
itself only polls every
+ // 50ms, so it would nearly always see a count that is added just after.
+ AtomicLong errorsAtRelease = new AtomicLong(-1);
+ LocalPipelineEngine run =
+ new LocalPipelineEngine(meta, new Variables(), new
LoggingObject("test")) {
+ @Override
+ public void pipelineCompleted() throws HopException {
+ errorsAtRelease.set(getResult().getNrErrors());
+ super.pipelineCompleted();
+ }
+ };
+ run.prepareExecution();
+ run.startThreads();
+ run.waitUntilFinished();
+
+ assertEquals(1, errorsAtRelease.get());
+ assertEquals(1, run.getResult().getNrErrors());
+ } finally {
+ ExtensionPointMap.getInstance().removeExtensionPoint(pipelineCompleted);
+ }
+ }
+
+ /**
+ * A single threaded run has to announce its transforms like any other.
Those transforms are run
+ * by the executor instead of by threads of their own, and that pipeline
type used to fire neither
+ * TransformBeforeStart nor TransformFinished - so a single threaded run
produced no transform
+ * level lineage at all.
+ */
+ @Test
+ void singleThreadedPipelineAnnouncesItsTransforms() throws Exception {
+ List<String> order = Collections.synchronizedList(new ArrayList<>());
+
+ TestExtensionPointPlugin transformStart =
+ registerExtensionPoint(
+ HopExtensionPoint.TransformBeforeStart.id, () ->
order.add("transform-start"));
+ TestExtensionPointPlugin transformFinished =
+ registerExtensionPoint(
+ HopExtensionPoint.TransformFinished.id, () ->
order.add("transform-finished"));
+
+ try {
+ PipelineMeta meta = new PipelineMeta();
+ meta.setName("single threaded test");
+ meta.addTransform(new TransformMeta("A", new DummyMeta()));
+
+ LocalPipelineEngine run =
+ new LocalPipelineEngine(meta, new Variables(), new
LoggingObject("test"));
+ run.setPipelineType(PipelineMeta.PipelineType.SingleThreaded);
+ run.prepareExecution();
+ run.startThreads();
+
+ // What SingleThreadedPipelineEngine does with the executor...
+ SingleThreadedPipelineExecutor executor = new
SingleThreadedPipelineExecutor(run);
+ assertTrue(executor.init());
+ executor.oneIteration();
+ executor.dispose();
+
+ assertEquals(List.of("transform-start", "transform-finished"), order);
+ } finally {
+ ExtensionPointMap.getInstance().removeExtensionPoint(transformStart);
+ ExtensionPointMap.getInstance().removeExtensionPoint(transformFinished);
+ }
+ }
+
+ /**
+ * Hand the extension point map a fake extension point, so a test can watch
for it. The caller
+ * removes the returned plugin again.
+ */
+ private TestExtensionPointPlugin registerExtensionPoint(
+ String extensionPointId, Runnable action) {
+ IExtensionPoint<Object> extensionPoint = (logChannel, variables, object)
-> action.run();
+
+ TestExtensionPointPlugin plugin = mock(TestExtensionPointPlugin.class);
+ // For an extension point plugin the name is the extension point it
listens to, and the ids are
+ // the plugin's own ids: ExtensionPointPluginType extracts them that way.
+ when(plugin.getName()).thenReturn(extensionPointId);
+ when(plugin.getIds()).thenReturn(new String[] {"test-" +
extensionPointId});
+ when(plugin.getMainType()).thenReturn((Class) IExtensionPoint.class);
+ when(plugin.loadClass(IExtensionPoint.class)).thenReturn(extensionPoint);
+ ExtensionPointMap.getInstance().addExtensionPoint(plugin);
+ return plugin;
+ }
+
@Test
void testFinishStatus() throws Exception {
while (pipeline.isRunning()) {
@@ -291,6 +528,12 @@ class PipelineTest {
private final IExecutionFinishedListener<IPipelineEngine<PipelineMeta>>
listener = p -> {};
+ /**
+ * A plugin that loads its own classes, so a test can hand the extension
point map a fake
+ * extension point without going through the plugin registry.
+ */
+ private interface TestExtensionPointPlugin extends IClassLoadingPlugin,
IPlugin {}
+
private final IExecutionStoppedListener<IPipelineEngine<PipelineMeta>>
pipelineStoppedListener =
p -> {};
