This is an automated email from the ASF dual-hosted git repository.
caishunfeng 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 57825589f9 Fix nextBranch is empty in switch task (#17054)
57825589f9 is described below
commit 57825589f9df19d71f6d17cd50e749aa41031f95
Author: Wenjun Ruan <[email protected]>
AuthorDate: Fri Mar 14 13:45:52 2025 +0800
Fix nextBranch is empty in switch task (#17054)
Co-authored-by: caishunfeng <[email protected]>
---
.../plugin/switchtask/SwitchLogicTask.java | 3 +
.../master/engine/graph/SuccessorFlowAdjuster.java | 4 +-
.../integration/cases/WorkflowStartTestCase.java | 86 ++++++++++++++++++
...kflow_with_one_failed_switch_two_fake_task.yaml | 101 +++++++++++++++++++++
...flow_with_one_success_switch_two_fake_task.yaml | 101 +++++++++++++++++++++
5 files changed, 293 insertions(+), 2 deletions(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/switchtask/SwitchLogicTask.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/switchtask/SwitchLogicTask.java
index 60c78d7d5a..71dd3854d6 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/switchtask/SwitchLogicTask.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/switchtask/SwitchLogicTask.java
@@ -117,6 +117,9 @@ public class SwitchLogicTask extends
AbstractLogicTask<SwitchParameters> {
if (nextBranch == null) {
log.info("All switch item is not satisfied");
moveToDefaultBranch();
+ } else {
+ log.info("The condition is satisfied, move to the next branch:
{}", getTaskName(nextBranch));
+ taskParameters.setNextBranch(nextBranch);
}
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/SuccessorFlowAdjuster.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/SuccessorFlowAdjuster.java
index 01594d105f..1b50b2683e 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/SuccessorFlowAdjuster.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/SuccessorFlowAdjuster.java
@@ -119,7 +119,7 @@ public class SuccessorFlowAdjuster {
needSkippedBranch.add(switchResultVo.getNextNode());
}
}
- needSkippedBranch.remove(switchResult.getNextNode());
+ needSkippedBranch.remove(switchParameters.getNextBranch());
markTaskSkipped(taskExecutionRunnable, needSkippedBranch);
}
@@ -136,7 +136,7 @@ public class SuccessorFlowAdjuster {
taskExecutionRunnable.getWorkflowInstance().getName());
continue;
}
- workflowExecutionGraph.markTaskSkipped(taskExecutionRunnable);
+ workflowExecutionGraph.markTaskSkipped(branch);
}
}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
index 19650f8db7..0d180e2f7f 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
@@ -183,6 +183,92 @@ public class WorkflowStartTestCase extends
AbstractMasterIntegrationTestCase {
masterContainer.assertAllResourceReleased();
}
+ @Test
+ @DisplayName("Test start a workflow with one success switch task and two
fake task")
+ public void testStartWorkflow_with_oneSuccessSwitch_twoFakeTask() {
+ final String yaml =
"/it/start/workflow_with_one_success_switch_two_fake_task.yaml";
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+ final WorkflowDefinition parentWorkflow = context.getOneWorkflow();
+
+ final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO =
WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(parentWorkflow)
+ .runWorkflowCommandParam(new RunWorkflowCommandParam())
+ .build();
+ final Integer workflowInstanceId =
workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+ await()
+ .atMost(Duration.ofMinutes(1))
+ .untilAsserted(() -> {
+
+ Assertions
+
.assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .matches(
+ workflowInstance ->
workflowInstance.getState() == WorkflowExecutionStatus.SUCCESS)
+ .matches(
+ workflowInstance ->
workflowInstance.getIsSubWorkflow() == Flag.NO)
+ .matches(
+ workflowInstance ->
workflowInstance.getDryRun() == Flag.NO.getCode());
+
+ Assertions
+
.assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .hasSize(2)
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("switch_task");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ })
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("success_branch");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ });
+
+ });
+
+ masterContainer.assertAllResourceReleased();
+ }
+
+ @Test
+ @DisplayName("Test start a workflow with one failed switch task and two
fake task")
+ public void testStartWorkflow_with_oneFailedSwitch_twoFakeTask() {
+ final String yaml =
"/it/start/workflow_with_one_failed_switch_two_fake_task.yaml";
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+ final WorkflowDefinition parentWorkflow = context.getOneWorkflow();
+
+ final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO =
WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(parentWorkflow)
+ .runWorkflowCommandParam(new RunWorkflowCommandParam())
+ .build();
+ final Integer workflowInstanceId =
workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+ await()
+ .atMost(Duration.ofMinutes(1))
+ .untilAsserted(() -> {
+
+ Assertions
+
.assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .matches(
+ workflowInstance ->
workflowInstance.getState() == WorkflowExecutionStatus.SUCCESS)
+ .matches(
+ workflowInstance ->
workflowInstance.getIsSubWorkflow() == Flag.NO)
+ .matches(
+ workflowInstance ->
workflowInstance.getDryRun() == Flag.NO.getCode());
+
+ Assertions
+
.assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .hasSize(2)
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("switch_task");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ })
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("default_branch");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ });
+
+ });
+
+ masterContainer.assertAllResourceReleased();
+ }
+
@Test
@DisplayName("Test start a workflow with one sub workflow task(A) success")
public void testStartWorkflow_with_subWorkflowTask_success() {
diff --git
a/dolphinscheduler-master/src/test/resources/it/start/workflow_with_one_failed_switch_two_fake_task.yaml
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_one_failed_switch_two_fake_task.yaml
new file mode 100644
index 0000000000..4ba78d93da
--- /dev/null
+++
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_one_failed_switch_two_fake_task.yaml
@@ -0,0 +1,101 @@
+#
+# 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.
+#
+
+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_one_failed_switch_two_fake_task
+ code: 1
+ version: 1
+ projectCode: 1
+ description: This is a workflow with one switch task falied
+ releaseState: ONLINE
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ userId: 1
+ executionType: PARALLEL
+
+tasks:
+ - name: switch_task
+ code: 1
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: SWITCH
+ taskParams:
'{"localParams":[],"rawScript":"","resourceList":[],"switchResult":{"dependTaskList":[{"condition":"false
== true","nextNode":2}],"nextNode":3}}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: success_branch
+ code: 2
+ 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: default_branch
+ code: 3
+ 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: 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: 1
+ preTaskVersion: 1
+ postTaskCode: 3
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
diff --git
a/dolphinscheduler-master/src/test/resources/it/start/workflow_with_one_success_switch_two_fake_task.yaml
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_one_success_switch_two_fake_task.yaml
new file mode 100644
index 0000000000..c7e7f905b8
--- /dev/null
+++
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_one_success_switch_two_fake_task.yaml
@@ -0,0 +1,101 @@
+#
+# 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.
+#
+
+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_one_success_switch_two_fake_task
+ code: 1
+ version: 1
+ projectCode: 1
+ description: This is a workflow with one switch task success
+ releaseState: ONLINE
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ userId: 1
+ executionType: PARALLEL
+
+tasks:
+ - name: switch_task
+ code: 1
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: SWITCH
+ taskParams:
'{"localParams":[],"rawScript":"","resourceList":[],"switchResult":{"dependTaskList":[{"condition":"true
== true","nextNode":2}],"nextNode":3}}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: success_branch
+ code: 2
+ 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: default_branch
+ code: 3
+ 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: 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: 1
+ preTaskVersion: 1
+ postTaskCode: 3
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
