This is an automated email from the ASF dual-hosted git repository.
SbloodyS pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 1f71cfcbd9 [Fix-18406] Fix task instance failed but workflow instance
remains in running state while using continue strategy with a blocked
intermediate predecessor (#18415)
1f71cfcbd9 is described below
commit 1f71cfcbd9f983ccaed48f2ffeb849f7e0eb2fd6
Author: xiangzihao <[email protected]>
AuthorDate: Wed Jul 15 17:55:11 2026 +0800
[Fix-18406] Fix task instance failed but workflow instance remains in
running state while using continue strategy with a blocked intermediate
predecessor (#18415)
---
.../statemachine/AbstractWorkflowStateAction.java | 10 +-
.../cases/WorkflowStartGraphTestCase.java | 40 ++++++
...redecessor_using_failure_strategy_continue.yaml | 153 +++++++++++++++++++++
3 files changed, 202 insertions(+), 1 deletion(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java
index 48b329bc77..b243b7b1e0 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java
@@ -94,7 +94,15 @@ public abstract class AbstractWorkflowStateAction implements
IWorkflowStateActio
.getPredecessors(taskExecution.getName())
.stream())
.allMatch(workflowExecutionGraph::isTaskExecutionInActive);
- if (isAllCandidateTaskPredecessorsInActive) {
+ final boolean isAllTaskExecutionChainFinish =
workflowExecutionGraph.isAllTaskExecutionChainFinish();
+ if (isAllCandidateTaskPredecessorsInActive ||
isAllTaskExecutionChainFinish) {
+ log.debug("No candidate task can be triggered in workflow: {},
check whether the workflow can finish, "
+ + "allCandidateTaskPredecessorsInActive: {},
allTaskExecutionChainFinish: {}, "
+ + "candidateTasks: {}",
+ workflowExecution.getName(),
+ isAllCandidateTaskPredecessorsInActive,
+ isAllTaskExecutionChainFinish,
+
triggerCandidateTasks.stream().map(ITaskExecution::getName).collect(Collectors.toList()));
emitWorkflowFinishedEventIfApplicable(workflowExecution);
}
return;
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartGraphTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartGraphTestCase.java
index c3b9137e43..c105112457 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartGraphTestCase.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartGraphTestCase.java
@@ -209,6 +209,46 @@ public class WorkflowStartGraphTestCase extends
AbstractMasterIntegrationTestCas
masterContainer.assertAllResourceReleased();
}
+ @Test
+ @DisplayName("Test continue failure strategy with a blocked intermediate
predecessor")
+ void
testStartWorkflow_with_blockedIntermediatePredecessor_usingFailureStrategyContinue()
{
+ final String yaml =
+
"/it/start/workflow_with_blocked_intermediate_predecessor_using_failure_strategy_continue.yaml";
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+ final WorkflowDefinition workflow = context.getOneWorkflow();
+
+ final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO =
WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(workflow)
+ .runWorkflowCommandParam(new RunWorkflowCommandParam())
+ .failureStrategy(FailureStrategy.CONTINUE)
+ .build();
+ final Integer workflowInstanceId =
workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+ await()
+ .atMost(Duration.ofMinutes(1))
+ .untilAsserted(() -> {
+
Assertions.assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .matches(workflowInstance ->
workflowInstance.getState() == WorkflowExecutionStatus.FAILURE)
+ .matches(workflowInstance ->
workflowInstance.getEndTime() != null);
+
+
Assertions.assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .hasSize(3)
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("A");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ })
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("B");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ })
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("C");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.FAILURE);
+ });
+ });
+ masterContainer.assertAllResourceReleased();
+ }
+
@Test
@DisplayName("Test start a workflow with two parallel fake tasks(A, B)
success")
public void testStartWorkflow_with_twoParallelSuccessTask() {
diff --git
a/dolphinscheduler-master/src/test/resources/it/start/workflow_with_blocked_intermediate_predecessor_using_failure_strategy_continue.yaml
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_blocked_intermediate_predecessor_using_failure_strategy_continue.yaml
new file mode 100644
index 0000000000..87ce533df0
--- /dev/null
+++
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_blocked_intermediate_predecessor_using_failure_strategy_continue.yaml
@@ -0,0 +1,153 @@
+#
+# 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.
+#
+
+# A(success) -> B(success) -> E(not triggered)
+# C(failed) -> D(not triggered) -> E(not triggered)
+# C finishes before A.
+project:
+ name: MasterIntegrationTest
+ code: 1
+ description: This is a fake project
+ userId: 1
+ userName: admin
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+
+workflows:
+ - name:
workflow_with_blocked_intermediate_predecessor_using_failure_strategy_continue
+ code: 1
+ version: 1
+ projectCode: 1
+ description: This is a fake workflow with a blocked intermediate
predecessor
+ releaseState: ONLINE
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ userId: 1
+ executionType: PARALLEL
+
+tasks:
+ - name: A
+ code: 1
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"sleep 1"}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: B
+ code: 2
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"sleep 1"}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: C
+ code: 3
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"xx"}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: D
+ code: 4
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"echo
success"}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: E
+ code: 5
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"echo
success"}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+
+taskRelations:
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 0
+ preTaskVersion: 0
+ postTaskCode: 1
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 0
+ preTaskVersion: 0
+ postTaskCode: 3
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 1
+ preTaskVersion: 1
+ postTaskCode: 2
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 3
+ preTaskVersion: 1
+ postTaskCode: 4
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 2
+ preTaskVersion: 1
+ postTaskCode: 5
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 4
+ preTaskVersion: 1
+ postTaskCode: 5
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
