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 ee4a3de3e72..737fd560b2b 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 @@ 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() + " : " @@ -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); @@ -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; @@ -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. + * + *

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 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 badGuys = new ArrayList<>(executionFinishedListeners.size()); - for (IExecutionFinishedListener> 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 badGuys = new ArrayList<>(executionFinishedListeners.size()); + for (IExecutionFinishedListener> 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 e0655d6a206..a45003a9f38 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.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 static void main(String[] args) { System.exit(1); } else { hopRun.run(); + + shutdownLineage(); + System.out.println("HopRun exit."); if (hopRun.isFinishedWithoutError()) { System.exit(0); @@ -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 // 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 4027a1fe51e..84ac02e2744 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.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 @@ void testFireExecutionFinishedListenersExceptionOnPipelineFinished() throws Exce 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 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 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 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 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 @@ public void run() { private final IExecutionFinishedListener> 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> pipelineStoppedListener = p -> {};