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 f3c8f2f887 Fix Join hanging when a parallel branch fails before
reaching Join (#8082)
f3c8f2f887 is described below
commit f3c8f2f88730468fc340ef4a48530f553c958589
Author: Lance <[email protected]>
AuthorDate: Tue Aug 25 21:43:12 2026 +0800
Fix Join hanging when a parallel branch fails before reaching Join (#8082)
Signed-off-by: lance <[email protected]>
---
.../0012-join-failure-with-intermediate.hwf | 168 +++++++++++++++++
integration-tests/actions/main-0012-join.hwf | 33 +++-
.../hop/workflow/actions/join/ActionJoin.java | 154 ++++++++++++++--
.../join/messages/messages_en_US.properties | 1 +
.../hop/workflow/actions/join/ActionJoinTest.java | 203 ++++++++++++++++++++-
5 files changed, 543 insertions(+), 16 deletions(-)
diff --git a/integration-tests/actions/0012-join-failure-with-intermediate.hwf
b/integration-tests/actions/0012-join-failure-with-intermediate.hwf
new file mode 100644
index 0000000000..b6a93e2f12
--- /dev/null
+++ b/integration-tests/actions/0012-join-failure-with-intermediate.hwf
@@ -0,0 +1,168 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<workflow>
+ <name>0012-join-failure-with-intermediate</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <created_user>-</created_user>
+ <modified_user>-</modified_user>
+ <created_date>2026/08/24 00:00:00.000</created_date>
+ <modified_date>2026/08/24 00:00:00.000</modified_date>
+ <workflow_version/>
+ <parameters/>
+ <actions>
+ <action>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <intervalSeconds>0</intervalSeconds>
+ <intervalMinutes>60</intervalMinutes>
+ <DayOfMonth>1</DayOfMonth>
+ <weekDay>1</weekDay>
+ <minutes>0</minutes>
+ <hour>12</hour>
+ <doNotWaitOnFirstExecution>N</doNotWaitOnFirstExecution>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <xloc>16</xloc>
+ <yloc>80</yloc>
+ <parallel>Y</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Join</name>
+ <description/>
+ <type>JOIN</type>
+ <attributes/>
+ <xloc>480</xloc>
+ <yloc>80</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Success</name>
+ <description/>
+ <type>SUCCESS</type>
+ <attributes/>
+ <xloc>608</xloc>
+ <yloc>80</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <maximumTimeout>1</maximumTimeout>
+ <scaletime>0</scaletime>
+ <name>Wait</name>
+ <description/>
+ <type>DELAY</type>
+ <attributes/>
+ <xloc>144</xloc>
+ <yloc>128</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <filename>NONE</filename>
+ <name>File exists</name>
+ <description/>
+ <type>FILE_EXISTS</type>
+ <attributes/>
+ <xloc>144</xloc>
+ <yloc>32</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Dummy after fail</name>
+ <description/>
+ <type>DUMMY</type>
+ <attributes/>
+ <xloc>272</xloc>
+ <yloc>32</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Dummy after wait</name>
+ <description/>
+ <type>DUMMY</type>
+ <attributes/>
+ <xloc>272</xloc>
+ <yloc>128</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>Wait</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Join</from>
+ <to>Success</to>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Start</from>
+ <to>File exists</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>File exists</from>
+ <to>Dummy after fail</to>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Dummy after fail</from>
+ <to>Join</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Wait</from>
+ <to>Dummy after wait</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Dummy after wait</from>
+ <to>Join</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ </hops>
+ <notepads/>
+ <attributes/>
+</workflow>
diff --git a/integration-tests/actions/main-0012-join.hwf
b/integration-tests/actions/main-0012-join.hwf
index 010716c6ed..da3ebd77bb 100644
--- a/integration-tests/actions/main-0012-join.hwf
+++ b/integration-tests/actions/main-0012-join.hwf
@@ -80,7 +80,7 @@ limitations under the License.
<attributes/>
<parallel>N</parallel>
<xloc>176</xloc>
- <yloc>512</yloc>
+ <yloc>624</yloc>
<attributes_hac/>
</action>
<action>
@@ -133,6 +133,30 @@ limitations under the License.
<yloc>400</yloc>
<attributes_hac/>
</action>
+ <action>
+ <name>0012-join-failure-with-intermediate.hwf</name>
+ <description/>
+ <type>WORKFLOW</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+
<filename>${PROJECT_HOME}/0012-join-failure-with-intermediate.hwf</filename>
+ <loglevel>Nothing</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>176</xloc>
+ <yloc>512</yloc>
+ <attributes_hac/>
+ </action>
</actions>
<hops>
<hop>
@@ -158,6 +182,13 @@ limitations under the License.
</hop>
<hop>
<from>0012-join-failure-with-true.hwf</from>
+ <to>0012-join-failure-with-intermediate.hwf</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0012-join-failure-with-intermediate.hwf</from>
<to>Success</to>
<enabled>Y</enabled>
<evaluation>N</evaluation>
diff --git
a/plugins/actions/join/src/main/java/org/apache/hop/workflow/actions/join/ActionJoin.java
b/plugins/actions/join/src/main/java/org/apache/hop/workflow/actions/join/ActionJoin.java
index 45cbe83cb3..0bf568e67c 100644
---
a/plugins/actions/join/src/main/java/org/apache/hop/workflow/actions/join/ActionJoin.java
+++
b/plugins/actions/join/src/main/java/org/apache/hop/workflow/actions/join/ActionJoin.java
@@ -17,15 +17,21 @@
package org.apache.hop.workflow.actions.join;
+import java.time.Duration;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
+import org.apache.commons.lang3.ThreadUtils;
import org.apache.hop.core.CheckResult;
import org.apache.hop.core.ICheckResult;
import org.apache.hop.core.Result;
import org.apache.hop.core.annotations.Action;
+import org.apache.hop.core.gui.WorkflowTracker;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.workflow.ActionResult;
import org.apache.hop.workflow.WorkflowHopMeta;
import org.apache.hop.workflow.WorkflowMeta;
import org.apache.hop.workflow.action.ActionBase;
@@ -66,39 +72,46 @@ public class ActionJoin extends ActionBase {
@Override
public Result execute(Result result, int nr) {
try {
-
// Find previous actions to join
List<ActionMeta> prevActions = getPreviousAction(this, new
ArrayList<>(), false);
- var workflowTracker = this.parentWorkflow.getWorkflowTracker();
while (!parentWorkflow.isStopped()) {
- Thread.sleep(500L);
+ ThreadUtils.sleep(Duration.ofMillis(500L));
boolean completed = true;
boolean success = true;
int errors = 0;
- // Checks if all previous actions have completed successfully
+ // Checks if all previous actions have completed, or can never run
for (ActionMeta actionMeta : prevActions) {
- var tracker = workflowTracker.findWorkflowTracker(actionMeta);
- if (tracker != null) {
- Result actionResult = tracker.getActionResult().getResult();
- if (actionResult == null) {
- completed = false;
- } else if (!actionResult.isResult()) {
+ Result actionResult = getFinishedActionResult(actionMeta);
+ if (actionResult != null) {
+ if (!actionResult.isResult()) {
WorkflowHopMeta hopMeta = findWorkflowHop(actionMeta);
// If one previous action has failure and the hop is true
evaluation, repeat failure
// to the join action
- if (!hopMeta.isUnconditional() && hopMeta.isEvaluation()) {
+ if (hopMeta != null && !hopMeta.isUnconditional() &&
hopMeta.isEvaluation()) {
success = false;
errors++;
}
}
+ } else if (willNeverExecute(actionMeta, new HashSet<>())) {
+ // Predecessor was skipped because an upstream hop was not
followed (for example a
+ // failed action with only a success hop toward this branch). Do
not wait forever.
+ if (isUnreachableBecauseOfFailure(actionMeta, new HashSet<>())) {
+ success = false;
+ errors++;
+ }
+ if (isBasic()) {
+ logBasic(
+ BaseMessages.getString(
+ PKG, "ActionJoin.Log.PredecessorUnreachable",
actionMeta.getName()));
+ }
} else {
completed = false;
}
}
- // If all previous actions have a result
+ // If all previous actions have a result or can never execute
if (completed) {
result.setResult(success);
result.setNrErrors(errors);
@@ -106,6 +119,10 @@ public class ActionJoin extends ActionBase {
}
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+
result.setNrErrors(1);
result.setResult(false);
logError(BaseMessages.getString(PKG, "ActionJoin.Error.CouldNotExecute")
+ e);
@@ -170,6 +187,119 @@ public class ActionJoin extends ActionBase {
return null;
}
+ /**
+ * Result of a finished action, or {@code null} if it has not started or is
still running. Matches
+ * {@link org.apache.hop.workflow.Workflow} hop following: a tracker with a
null result is the
+ * "started" marker, not a completed execution.
+ */
+ private Result getFinishedActionResult(ActionMeta actionMeta) {
+ WorkflowTracker<?> tracker =
+ parentWorkflow.getWorkflowTracker().findWorkflowTracker(actionMeta);
+ if (tracker == null) {
+ return null;
+ }
+
+ ActionResult actionResult = tracker.getActionResult();
+ if (actionResult == null) {
+ return null;
+ }
+ return actionResult.getResult();
+ }
+
+ /**
+ * Same condition as {@link org.apache.hop.workflow.Workflow} when deciding
whether to execute the
+ * next action after {@code fromAction} finished with {@code fromResult}.
+ */
+ private static boolean isHopFollowed(
+ WorkflowHopMeta hop, ActionMeta fromAction, Result fromResult) {
+ return hop.isUnconditional()
+ || (fromAction.isEvaluation() && hop.isEvaluation() ==
fromResult.isResult());
+ }
+
+ /**
+ * True when {@code actionMeta} has not started and every enabled incoming
hop is dead: the
+ * previous action finished without following the hop, or that previous
action itself will never
+ * execute. Conservatively returns false if a predecessor is still running
or is about to start.
+ */
+ private boolean willNeverExecute(ActionMeta actionMeta, Set<ActionMeta>
visiting) {
+ if (getFinishedActionResult(actionMeta) != null) {
+ return false;
+ }
+ if (parentWorkflow.getWorkflowTracker().findWorkflowTracker(actionMeta) !=
null
+ || actionMeta.isStart()) {
+ return false;
+ }
+ if (!visiting.add(actionMeta)) {
+ return false;
+ }
+
+ List<WorkflowHopMeta> incoming = findIncomingHops(actionMeta);
+ if (incoming.isEmpty()) {
+ return true;
+ }
+
+ for (WorkflowHopMeta hop : incoming) {
+ ActionMeta fromAction = hop.getFromAction();
+ if (fromAction == null) {
+ continue;
+ }
+
+ Result fromResult = getFinishedActionResult(fromAction);
+ if (fromResult != null) {
+ if (isHopFollowed(hop, fromAction, fromResult)) {
+ return false;
+ }
+ } else if (!willNeverExecute(fromAction, visiting)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * True when the action is unreachable because a success hop was not
followed after a failure.
+ * False when it is unreachable because a failure hop was not followed after
a success, so Join
+ * should not fail.
+ */
+ private boolean isUnreachableBecauseOfFailure(ActionMeta actionMeta,
Set<ActionMeta> visiting) {
+ if (!visiting.add(actionMeta)) {
+ return false;
+ }
+ for (WorkflowHopMeta hop : findIncomingHops(actionMeta)) {
+ ActionMeta fromAction = hop.getFromAction();
+ if (fromAction == null) {
+ continue;
+ }
+ Result fromResult = getFinishedActionResult(fromAction);
+ if (fromResult != null) {
+ if (!isHopFollowed(hop, fromAction, fromResult)
+ && !fromResult.isResult()
+ && !hop.isUnconditional()
+ && hop.isEvaluation()) {
+ return true;
+ }
+ } else if (willNeverExecute(fromAction, new HashSet<>())
+ && isUnreachableBecauseOfFailure(fromAction, visiting)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private List<WorkflowHopMeta> findIncomingHops(ActionMeta toAction) {
+ List<WorkflowHopMeta> incoming = new ArrayList<>();
+ if (parentWorkflowMeta == null) {
+ return incoming;
+ }
+
+ for (WorkflowHopMeta hop : parentWorkflowMeta.getWorkflowHops()) {
+ if (hop.isEnabled() && hop.getToAction() != null &&
hop.getToAction().equals(toAction)) {
+ incoming.add(hop);
+ }
+ }
+ return incoming;
+ }
+
/** Find previous actions */
private List<ActionMeta> getPreviousAction(
IAction action, List<ActionMeta> prevActions, boolean deep) {
diff --git
a/plugins/actions/join/src/main/resources/org/apache/hop/workflow/actions/join/messages/messages_en_US.properties
b/plugins/actions/join/src/main/resources/org/apache/hop/workflow/actions/join/messages/messages_en_US.properties
index b73657a13e..f0b3670b89 100644
---
a/plugins/actions/join/src/main/resources/org/apache/hop/workflow/actions/join/messages/messages_en_US.properties
+++
b/plugins/actions/join/src/main/resources/org/apache/hop/workflow/actions/join/messages/messages_en_US.properties
@@ -21,5 +21,6 @@ ActionJoin.CheckResult.NoParallelExecution=No parallel
execution to join
ActionJoin.Description=Join branch of a workflow executed in parallel
ActionJoin.Error.CouldNotExecute=Couldn't execute join action
ActionJoin.Keyword=join,synchronize,merge,parallel,branch
+ActionJoin.Log.PredecessorUnreachable=Predecessor action ''{0}'' will never
execute; Join will not wait for it
ActionJoin.Name=Join
ActionJoinDialog.Title=Join
diff --git
a/plugins/actions/join/src/test/java/org/apache/hop/workflow/actions/join/ActionJoinTest.java
b/plugins/actions/join/src/test/java/org/apache/hop/workflow/actions/join/ActionJoinTest.java
index 3f753b2578..d1ccaf6197 100644
---
a/plugins/actions/join/src/test/java/org/apache/hop/workflow/actions/join/ActionJoinTest.java
+++
b/plugins/actions/join/src/test/java/org/apache/hop/workflow/actions/join/ActionJoinTest.java
@@ -23,19 +23,30 @@ import static
org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.ThreadUtils;
import org.apache.hop.core.Result;
+import org.apache.hop.core.logging.LogLevel;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.workflow.WorkflowHopMeta;
import org.apache.hop.workflow.WorkflowMeta;
+import org.apache.hop.workflow.action.ActionBase;
+import org.apache.hop.workflow.action.ActionMeta;
+import org.apache.hop.workflow.actions.dummy.ActionDummy;
+import org.apache.hop.workflow.actions.start.ActionStart;
import org.apache.hop.workflow.engine.IWorkflowEngine;
import org.apache.hop.workflow.engines.local.LocalWorkflowEngine;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
+/** Unit test for {@link ActionJoin} */
@ExtendWith(RestoreHopEngineEnvironmentExtension.class)
class ActionJoinTest {
@@ -116,7 +127,7 @@ class ActionJoinTest {
}
@Test
- void testExecuteWithNoPreviousActions() throws Exception {
+ void testExecuteWithNoPreviousActions() {
Result result = new Result();
Result executionResult = action.execute(result, 0);
@@ -125,7 +136,7 @@ class ActionJoinTest {
}
@Test
- void testExecuteWithPreviousActions() throws Exception {
+ void testExecuteWithPreviousActions() {
// Test execute with no previous actions (simplified test)
Result result = new Result();
Result executionResult = action.execute(result, 0);
@@ -135,7 +146,7 @@ class ActionJoinTest {
}
@Test
- void testExecuteWithException() throws Exception {
+ void testExecuteWithException() {
// Test execute with exception handling
Result result = new Result();
Result executionResult = action.execute(result, 0);
@@ -208,4 +219,190 @@ class ActionJoinTest {
// The method should execute without errors
assertNotNull(remarks);
}
+
+ @Test
+ @Timeout(value = 15, unit = TimeUnit.SECONDS)
+ void executeDoesNotHangWhenPredecessorNeverRunsAfterBranchFailure() {
+ WorkflowMeta meta = new WorkflowMeta();
+ meta.setName("join-unreachable-after-failure");
+
+ ActionMeta startMeta = new ActionMeta(new ActionStart("Start"));
+ startMeta.setLaunchingInParallel(true);
+ ActionMeta successMeta = new ActionMeta(new ActionDummy("Success branch"));
+ ActionMeta failMeta = new ActionMeta(new FailingEvalAction("Fail"));
+ ActionMeta afterFailMeta = new ActionMeta(new ActionDummy("After fail"));
+ ActionMeta joinMeta = new ActionMeta(new ActionJoin("Join", ""));
+
+ meta.addAction(startMeta);
+ meta.addAction(successMeta);
+ meta.addAction(failMeta);
+ meta.addAction(afterFailMeta);
+ meta.addAction(joinMeta);
+
+ meta.addWorkflowHop(new WorkflowHopMeta(startMeta, successMeta));
+ meta.addWorkflowHop(new WorkflowHopMeta(startMeta, failMeta));
+
+ WorkflowHopMeta successToJoin = new WorkflowHopMeta(successMeta, joinMeta);
+ successToJoin.setUnconditional();
+ meta.addWorkflowHop(successToJoin);
+
+ WorkflowHopMeta failToAfter = new WorkflowHopMeta(failMeta, afterFailMeta);
+ failToAfter.setConditional();
+ failToAfter.setEvaluation(true);
+ meta.addWorkflowHop(failToAfter);
+
+ WorkflowHopMeta afterToJoin = new WorkflowHopMeta(afterFailMeta, joinMeta);
+ afterToJoin.setUnconditional();
+ meta.addWorkflowHop(afterToJoin);
+
+ LocalWorkflowEngine engine = new LocalWorkflowEngine(meta);
+ engine.setLogLevel(LogLevel.MINIMAL);
+ Result result = engine.startExecution();
+
+ assertFalse(result.isResult());
+ assertTrue(result.getNrErrors() >= 1);
+ }
+
+ @Test
+ @Timeout(value = 15, unit = TimeUnit.SECONDS)
+ void executeSucceedsWhenPredecessorNeverRunsBecauseFailureHopWasSkipped() {
+ WorkflowMeta meta = new WorkflowMeta();
+ meta.setName("join-unreachable-after-success");
+
+ ActionMeta startMeta = new ActionMeta(new ActionStart("Start"));
+ startMeta.setLaunchingInParallel(true);
+ ActionMeta successMeta = new ActionMeta(new ActionDummy("Success branch"));
+ ActionMeta evalMeta = new ActionMeta(new SucceedingEvalAction("Eval
success"));
+ ActionMeta neverMeta = new ActionMeta(new ActionDummy("Never run"));
+ ActionMeta joinMeta = new ActionMeta(new ActionJoin("Join", ""));
+
+ meta.addAction(startMeta);
+ meta.addAction(successMeta);
+ meta.addAction(evalMeta);
+ meta.addAction(neverMeta);
+ meta.addAction(joinMeta);
+
+ meta.addWorkflowHop(new WorkflowHopMeta(startMeta, successMeta));
+ meta.addWorkflowHop(new WorkflowHopMeta(startMeta, evalMeta));
+
+ WorkflowHopMeta successToJoin = new WorkflowHopMeta(successMeta, joinMeta);
+ successToJoin.setUnconditional();
+ meta.addWorkflowHop(successToJoin);
+
+ WorkflowHopMeta evalToNever = new WorkflowHopMeta(evalMeta, neverMeta);
+ evalToNever.setConditional();
+ evalToNever.setEvaluation(false);
+ meta.addWorkflowHop(evalToNever);
+
+ WorkflowHopMeta neverToJoin = new WorkflowHopMeta(neverMeta, joinMeta);
+ neverToJoin.setUnconditional();
+ meta.addWorkflowHop(neverToJoin);
+
+ LocalWorkflowEngine engine = new LocalWorkflowEngine(meta);
+ engine.setLogLevel(LogLevel.MINIMAL);
+ Result result = engine.startExecution();
+
+ assertTrue(result.isResult());
+ assertEquals(0, result.getNrErrors());
+ }
+
+ @Test
+ @Timeout(value = 15, unit = TimeUnit.SECONDS)
+ void executeStillWaitsForSlowPredecessorThatHasNotStartedYet() {
+ WorkflowMeta meta = new WorkflowMeta();
+ meta.setName("join-wait-for-slow-branch");
+
+ ActionMeta startMeta = new ActionMeta(new ActionStart("Start"));
+ startMeta.setLaunchingInParallel(true);
+ ActionMeta fastMeta = new ActionMeta(new ActionDummy("Fast branch"));
+ ActionMeta slowMeta = new ActionMeta(new SleepingEvalAction("Slow branch",
800));
+ ActionMeta joinMeta = new ActionMeta(new ActionJoin("Join", ""));
+
+ meta.addAction(startMeta);
+ meta.addAction(fastMeta);
+ meta.addAction(slowMeta);
+ meta.addAction(joinMeta);
+
+ meta.addWorkflowHop(new WorkflowHopMeta(startMeta, fastMeta));
+ meta.addWorkflowHop(new WorkflowHopMeta(startMeta, slowMeta));
+
+ WorkflowHopMeta fastToJoin = new WorkflowHopMeta(fastMeta, joinMeta);
+ fastToJoin.setUnconditional();
+ meta.addWorkflowHop(fastToJoin);
+
+ WorkflowHopMeta slowToJoin = new WorkflowHopMeta(slowMeta, joinMeta);
+ slowToJoin.setUnconditional();
+ meta.addWorkflowHop(slowToJoin);
+
+ LocalWorkflowEngine engine = new LocalWorkflowEngine(meta);
+ engine.setLogLevel(LogLevel.MINIMAL);
+ Result result = engine.startExecution();
+
+ assertTrue(result.isResult());
+
assertNotNull(engine.getWorkflowTracker().findWorkflowTracker(slowMeta).getActionResult());
+ assertNotNull(
+
engine.getWorkflowTracker().findWorkflowTracker(slowMeta).getActionResult().getResult());
+ }
+
+ static class FailingEvalAction extends ActionBase {
+ FailingEvalAction(String name) {
+ super(name, "");
+ }
+
+ @Override
+ public Result execute(Result result, int nr) {
+ result.setResult(false);
+ result.setNrErrors(1);
+ return result;
+ }
+
+ @Override
+ public boolean isEvaluation() {
+ return true;
+ }
+ }
+
+ static class SucceedingEvalAction extends ActionBase {
+ SucceedingEvalAction(String name) {
+ super(name, "");
+ }
+
+ @Override
+ public Result execute(Result result, int nr) {
+ result.setResult(true);
+ result.setNrErrors(0);
+ return result;
+ }
+
+ @Override
+ public boolean isEvaluation() {
+ return true;
+ }
+ }
+
+ static class SleepingEvalAction extends ActionBase {
+ private final long sleepMs;
+
+ SleepingEvalAction(String name, long sleepMs) {
+ super(name, "");
+ this.sleepMs = sleepMs;
+ }
+
+ @Override
+ public Result execute(Result result, int nr) {
+ try {
+ ThreadUtils.sleep(Duration.ofMillis(sleepMs));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ result.setResult(true);
+ result.setNrErrors(0);
+ return result;
+ }
+
+ @Override
+ public boolean isEvaluation() {
+ return true;
+ }
+ }
}
