mattcasters commented on code in PR #9035:
URL: https://github.com/apache/hop/pull/9035#discussion_r4334089355
##########
engine/src/main/java/org/apache/hop/pipeline/Pipeline.java:
##########
@@ -1579,15 +1610,21 @@ public void fireExecutionFinishedListeners() throws
HopException {
}
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);
} finally {
+ // Now the status and everything else is set correctly. We've completed
the pipeline.
+ //
+ // Always release the waiters, even when an extension point failed: that
must never be able to
+ // deadlock a caller. The extension point failure itself still
propagates.
+ //
+ pipelineCompleted();
Review Comment:
**[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`.
##########
engine/src/test/java/org/apache/hop/pipeline/PipelineTest.java:
##########
@@ -150,6 +163,182 @@ 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<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(
+ () -> {
+ line.waitUntilFinished();
Review Comment:
**[bug]** This waiter calls `waitUntilFinished()` before `startThreads()`
creates `pipelineWaitUntilFinishedBlockingQueue`. That method returns
immediately while the queue is still null, so the waiter appends `"waiter"` and
counts the latch down before `PipelineCompleted` runs. `startThreads()` invokes
`PipelineStartThreads` and other extension points before allocating the queue,
so the new thread often wins that window and the assertion sees `[waiter,
extension-point]` even though the production order is correct. The 400ms wait
inside the extension point does not cover this case: the waiter has already
recorded itself.
**Suggestion:** Do not call `waitUntilFinished()` until the queue exists. It
is assigned before `setRunning(true)`, and an empty pipeline signals only after
that. Spin while `!line.isRunning() && !line.isFinished()`, then wait. A late
waiter still observes the object already on the queue, so the expected order
stays deterministic without depending on the null-queue return.
##########
engine/src/main/java/org/apache/hop/run/HopRun.java:
##########
@@ -92,6 +93,14 @@ public static void main(String[] args) {
System.exit(1);
} else {
hopRun.run();
+
+ // 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. Safe to call when
+ // lineage is disabled.
+ //
+ LineageHub.getInstance().shutdown();
Review Comment:
**[suggestion]** `LineageHub.shutdown()` runs only when `hopRun.run()`
returns. The `ExecutionException` and other catch paths call `System.exit`
without draining. `TransformBeforeStart` can now throw from `startThreads()`
after an earlier transform thread has already emitted, and the OpenLineage sink
drains its own dispatcher only from `shutdown()`.
**Suggestion:** Call `shutdown()` on those failure paths immediately before
`System.exit` as well. A `finally` attached to this try does not run once
`System.exit` has been entered from a catch. The call is already a no-op when
the hub never started.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
