[Fix-17453] Fix TASK_ONLY strategy cannot work

Author: ruanwenjun

dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/command/RunWorkflowCommandParam.java

@@ -19,6 +19,8 @@
 
 import org.apache.dolphinscheduler.common.enums.CommandType;
 
+import java.util.List;
+
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
@@ -35,4 +37,9 @@ public CommandType getCommandType() {
         return CommandType.START_PROCESS;
     }
 
+    public RunWorkflowCommandParam withStartNodes(List<Long> startNodes) {
+        this.startNodes = startNodes;
+        return this;
+    }
+
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RecoverFailureTaskCommandHandler.java

@@ -139,6 +139,7 @@ protected void assembleWorkflowExecutionGraph(final WorkflowExecuteContextBuilde
                         .doVisitFunction(taskExecutionRunnableCreator)
                         .build();
         workflowGraphTopologyLogicalVisitor.visit();
+        workflowExecutionGraph.removeUnReachableEdge();
 
         workflowExecuteContextBuilder.setWorkflowExecutionGraph(workflowExecutionGraph);
     }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RunWorkflowCommandHandler.java

@@ -116,6 +116,7 @@ protected void assembleWorkflowExecutionGraph(final WorkflowExecuteContextBuilde
                         .doVisitFunction(taskExecutionRunnableCreator)
                         .build();
         workflowGraphTopologyLogicalVisitor.visit();
+        workflowExecutionGraph.removeUnReachableEdge();
 
         workflowExecuteContextBuilder.setWorkflowExecutionGraph(workflowExecutionGraph);
     }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/WorkflowFailoverCommandHandler.java

@@ -137,6 +137,7 @@ protected void assembleWorkflowExecutionGraph(final WorkflowExecuteContextBuilde
                         .doVisitFunction(taskExecutionRunnableCreator)
                         .build();
         workflowGraphTopologyLogicalVisitor.visit();
+        workflowExecutionGraph.removeUnReachableEdge();
 
         workflowExecuteContextBuilder.setWorkflowExecutionGraph(workflowExecutionGraph);
     }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/IWorkflowExecutionGraph.java

@@ -40,6 +40,11 @@ public interface IWorkflowExecutionGraph {
      */
     void addEdge(final String fromTaskName, final Set<String> toTaskName);
 
+    /**
+     * Remove the unreachable edge in the graph.
+     */
+    void removeUnReachableEdge();
+
     /**
      * Return the start tasks, the start tasks in the workflow execution graph is the tasks which predecessors is empty.
      */

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowExecutionGraph.java

@@ -28,13 +28,16 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 public class WorkflowExecutionGraph implements IWorkflowExecutionGraph {
 
+    // Store all the task execution runnable in the execution graph.
     private final Map<String, ITaskExecutionRunnable> totalTaskExecuteRunnableMap;
 
     private final Set<String> failureTaskChains;
@@ -78,6 +81,26 @@ public void addEdge(String fromTaskName, Set<String> toTaskNames) {
         toTaskNames.forEach(toTask -> predecessors.computeIfAbsent(toTask, k -> new HashSet<>()).add(fromTaskName));
     }
 
+    @Override
+    public void removeUnReachableEdge() {
+        // If the node in successors or predecessors is not in taskExecuteRunnableMap
+        // It means that the node is not executable, so we need to filter it out
+        Consumer<Map<String, Set<String>>> removeUnReachableEdge = edgeMap -> {
+            final Iterator<Map.Entry<String, Set<String>>> iterator = edgeMap.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Map.Entry<String, Set<String>> entry = iterator.next();
+                if (!totalTaskExecuteRunnableMap.containsKey(entry.getKey())) {
+                    iterator.remove();
+                    continue;
+                }
+                Set<String> toTasks = entry.getValue();
+                toTasks.removeIf(toTask -> !totalTaskExecuteRunnableMap.containsKey(toTask));
+            }
+        };
+        removeUnReachableEdge.accept(successors);
+        removeUnReachableEdge.accept(predecessors);
+    }
+
     @Override
     public List<ITaskExecutionRunnable> getStartNodes() {
         return totalTaskExecuteRunnableMap.values()

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/WorkflowOperator.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.master.integration;
 
 import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.TaskDependType;
 import org.apache.dolphinscheduler.dao.entity.Project;
 import org.apache.dolphinscheduler.dao.entity.Schedule;
 import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
@@ -62,7 +63,8 @@ public Integer manualTriggerWorkflow(final WorkflowTriggerDTO workflowTriggerDTO
                 .workflowDefinitionVersion(workflowTriggerDTO.workflowDefinition.getVersion())
                 .startNodes(workflowTriggerDTO.getRunWorkflowCommandParam().getStartNodes())
                 .startParamList(workflowTriggerDTO.getRunWorkflowCommandParam().getCommandParams())
-                .dryRun(workflowTriggerDTO.dryRun)
+                .dryRun(workflowTriggerDTO.getDryRun())
+                .taskDependType(workflowTriggerDTO.getTaskDependType())
                 .build();
 
         final WorkflowManualTriggerResponse manualTriggerWorkflowResponse =
@@ -150,6 +152,9 @@ public static class WorkflowTriggerDTO {
 
         @Builder.Default
         private Flag dryRun = Flag.NO;
+
+        @Builder.Default
+        private TaskDependType taskDependType = TaskDependType.TASK_POST;
     }
 
     @Data

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java

@@ -21,6 +21,7 @@
 import static org.awaitility.Awaitility.await;
 
 import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.TaskDependType;
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
@@ -1035,4 +1036,36 @@ public void testStartWorkflow_withTimeoutKillTask() {
                 });
         masterContainer.assertAllResourceReleased();
     }
+
+    @Test
+    @DisplayName("Test start a workflow with task depend type TASK_ONLY")
+    public void testStartWorkflow_withTaskOnlyStrategy() {
+        final String yaml = "/it/start/workflow_with_task_only_strategy.yaml";
+        final WorkflowTestCaseContext context = workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+        final WorkflowDefinition workflow = context.getOneWorkflow();
+
+        final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO = WorkflowOperator.WorkflowTriggerDTO.builder()
+                .workflowDefinition(workflow)
+                .runWorkflowCommandParam(new RunWorkflowCommandParam().withStartNodes(Lists.newArrayList(1L)))
+                .taskDependType(TaskDependType.TASK_ONLY)
+                .build();
+        workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    Assertions
+                            .assertThat(repository.queryWorkflowInstance(workflow))
+                            .satisfiesExactly(workflowInstance -> assertThat(workflowInstance.getState())
+                                    .isEqualTo(WorkflowExecutionStatus.SUCCESS));
+                    Assertions
+                            .assertThat(repository.queryTaskInstance(workflow))
+                            .hasSize(1)
+                            .satisfiesExactly(taskInstance -> {
+                                assertThat(taskInstance.getName()).isEqualTo("A");
+                                assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+                            });
+                });
+        masterContainer.assertAllResourceReleased();
+    }
 }

dolphinscheduler-master/src/test/resources/it/start/workflow_with_task_only_strategy.yaml

@@ -0,0 +1,80 @@
+#
+# 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_two_serial_fake_task_success
+    code: 1
+    version: 1
+    projectCode: 1
+    description: This is a fake workflow with two serial tasks
+    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":"echo hello"}'
+    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":"echo hello"}'
+    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