[3.1.8] [fix-#14537] the branch that needs to be executed overlaps with another branch, it may not be able to complete the normal execution #14563 (#14689)

Co-authored-by: fuchanghai <33984497+fuchanghai@users.noreply.github.com>

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/utils/DagHelper.java

@@ -413,28 +413,74 @@ public class DagHelper {
         return conditionTaskList;
     }
 
-    private static List<String> skipTaskNode4Switch(TaskNode taskNode, Map<String, TaskNode> skipTaskNodeList,
-                                                    Map<String, TaskInstance> completeTaskList,
-                                                    DAG<String, TaskNode, TaskNodeRelation> dag) {
+
+    public static List<String> skipTaskNode4Switch(TaskNode taskNode,
+                                                 Map<String, TaskNode> skipTaskNodeList,
+                                                 Map<String, TaskInstance> completeTaskList,
+                                                 DAG<String, TaskNode, TaskNodeRelation> dag) {
 
         SwitchParameters switchParameters =
                 completeTaskList.get(Long.toString(taskNode.getCode())).getSwitchDependency();
         int resultConditionLocation = switchParameters.getResultConditionLocation();
         List<SwitchResultVo> conditionResultVoList = switchParameters.getDependTaskList();
+
         List<String> switchTaskList = conditionResultVoList.get(resultConditionLocation).getNextNode();
+        Set<String> switchNeedWorkCodes = new HashSet<>();
         if (CollectionUtils.isEmpty(switchTaskList)) {
-            switchTaskList = new ArrayList<>();
+            return new ArrayList<>();
+        }
+        // get all downstream nodes of the branch that the switch node needs to execute
+        for (String switchTaskCode : switchTaskList) {
+            getSwitchNeedWorkCodes(switchTaskCode, dag, switchNeedWorkCodes);
         }
         conditionResultVoList.remove(resultConditionLocation);
         for (SwitchResultVo info : conditionResultVoList) {
             if (CollectionUtils.isEmpty(info.getNextNode())) {
                 continue;
             }
-            setTaskNodeSkip(info.getNextNode().get(0), dag, completeTaskList, skipTaskNodeList);
+            for (String nextNode : info.getNextNode()) {
+                setSwitchTaskNodeSkip(nextNode, dag, completeTaskList, skipTaskNodeList,
+                        switchNeedWorkCodes);
+            }
         }
         return switchTaskList;
     }
 
+    /**
+     * get all downstream nodes of the branch that the switch node needs to execute
+     * @param taskCode
+     * @param dag
+     * @param switchNeedWorkCodes
+     */
+    public static void getSwitchNeedWorkCodes(String taskCode, DAG<String, TaskNode, TaskNodeRelation> dag,
+                                              Set<String> switchNeedWorkCodes) {
+        switchNeedWorkCodes.add(taskCode);
+        Set<String> subsequentNodes = dag.getSubsequentNodes(taskCode);
+        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(subsequentNodes)) {
+            for (String subCode : subsequentNodes) {
+                getSwitchNeedWorkCodes(subCode, dag, switchNeedWorkCodes);
+            }
+        }
+    }
+
+    private static void setSwitchTaskNodeSkip(String skipNodeCode,
+                                              DAG<String, TaskNode, TaskNodeRelation> dag,
+                                              Map<String, TaskInstance> completeTaskList,
+                                              Map<String, TaskNode> skipTaskNodeList,
+                                              Set<String> switchNeedWorkCodes) {
+        // ignore when the node that needs to be skipped exists on the branch that the switch type node needs to execute
+        if (!dag.containsNode(skipNodeCode) || switchNeedWorkCodes.contains(skipNodeCode)) {
+            return;
+        }
+        skipTaskNodeList.putIfAbsent(skipNodeCode, dag.getNode(skipNodeCode));
+        Collection<String> postNodeList = dag.getSubsequentNodes(skipNodeCode);
+        for (String post : postNodeList) {
+            TaskNode postNode = dag.getNode(post);
+            if (isTaskNodeNeedSkip(postNode, skipTaskNodeList)) {
+                setTaskNodeSkip(post, dag, completeTaskList, skipTaskNodeList);
+            }
+        }
+    }
     /**
      * set task node and the post nodes skip flag
      */

dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/utils/DagHelperTest.java

@@ -330,6 +330,24 @@ public class DagHelperTest {
         Assert.assertEquals(1, postNodes.size());
     }
 
+    @Test
+    public void testSwitchPostNode() throws IOException {
+        DAG<String, TaskNode, TaskNodeRelation> dag = generateDag2();
+        Map<String, TaskNode> skipTaskNodeList = new HashMap<>();
+        Map<String, TaskInstance> completeTaskList = new HashMap<>();
+        completeTaskList.put("0", new TaskInstance());
+        TaskInstance taskInstance = new TaskInstance();
+        taskInstance.setState(TaskExecutionStatus.SUCCESS);
+        taskInstance.setTaskCode(1);
+        Map<String, Object> taskParamsMap = new HashMap<>();
+        taskParamsMap.put(Constants.SWITCH_RESULT, "");
+        taskInstance.setTaskParams(JSONUtils.toJsonString(taskParamsMap));
+        taskInstance.setSwitchDependency(getSwitchNode());
+        completeTaskList.put("1", taskInstance);
+        DagHelper.skipTaskNode4Switch(dag.getNode("1"), skipTaskNodeList, completeTaskList, dag);
+        Assert.assertNotNull(skipTaskNodeList.get("2"));
+        Assert.assertEquals(1, skipTaskNodeList.size());
+    }
     /**
      * process:
      * 1->2->3->5->7
@@ -436,11 +454,13 @@ public class DagHelperTest {
 
     /**
      * DAG graph:
-     *    2
-     *    ↑
-     * 0->1(switch)
-     *    ↓
-     *    4
+     *           -> 2->
+     *         /        \
+     *      /               \
+     * 0->1(switch)->5          6
+     *      \               /
+     *        \         /
+     *          -> 4->
      *
      * @return dag
      * @throws JsonProcessingException if error throws JsonProcessingException
@@ -484,15 +504,26 @@ public class DagHelperTest {
         taskNodeList.add(node4);
 
         TaskNode node5 = new TaskNode();
-        node5.setId("4");
-        node5.setName("4");
-        node5.setCode(4);
+        node5.setId("5");
+        node5.setName("5");
+        node5.setCode(5);
         node5.setType("SHELL");
         List<String> dep5 = new ArrayList<>();
         dep5.add("1");
         node5.setPreTasks(JSONUtils.toJsonString(dep5));
         taskNodeList.add(node5);
 
+        TaskNode node6 = new TaskNode();
+        node5.setId("6");
+        node5.setName("6");
+        node5.setCode(6);
+        node5.setType("SHELL");
+        List<String> dep6 = new ArrayList<>();
+        dep5.add("2");
+        dep5.add("4");
+        node5.setPreTasks(JSONUtils.toJsonString(dep6));
+        taskNodeList.add(node6);
+
         List<String> startNodes = new ArrayList<>();
         List<String> recoveryNodes = new ArrayList<>();
         List<TaskNode> destTaskNodeList = DagHelper.generateFlowNodeListByStartNode(taskNodeList,
@@ -518,7 +549,7 @@ public class DagHelperTest {
         conditionsParameters.setDependTaskList(list);
         conditionsParameters.setNextNode("5");
         conditionsParameters.setRelation("AND");
-
+        conditionsParameters.setResultConditionLocation(1);
         // in: AND(AND(1 is SUCCESS))
         return conditionsParameters;
     }