You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hop.apache.org by "bamaer (via GitHub)" <gi...@apache.org> on 2026/10/20 11:03:54 UTC
[PR] Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005 (hop)
bamaer opened a new pull request, #9035:
URL: https://github.com/apache/hop/pull/9035
- 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
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [x] Run `mvn clean install apache-rat:check` to make sure basic checks pass. A more thorough check will be performed on your pull request automatically.
- [x] If you have a group of commits related to the same change, please squash your commits into one and force push your branch using `git rebase -i`.
- [x] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable.
To make clear that you license your contribution under the [Apache License Version 2.0, January 2004](http://www.apache.org/licenses/LICENSE-2.0)
you have to acknowledge this by using the following check-box.
- [x] I hereby declare this contribution to be licensed under the [Apache License Version 2.0, January 2004](http://www.apache.org/licenses/LICENSE-2.0)
- [ ] In any other case, please file an [Apache Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
--
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: issues-unsubscribe@hop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005 (hop)
Posted by "mattcasters (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@hop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Run PipelineCompleted before releasing waitUntilFinished(), fixes #9005 (hop)
Posted by "hansva (via GitHub)" <gi...@apache.org>.
hansva merged PR #9035:
URL: https://github.com/apache/hop/pull/9035
--
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: issues-unsubscribe@hop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org