[Fix-17350] Fix issues of sub-workflow failover from different master server (#17352)

Author: reele

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java

@@ -17,7 +17,6 @@
 
 package org.apache.dolphinscheduler.common.enums;
 
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -94,8 +93,19 @@ public boolean canPause() {
                 || this == SERIAL_WAIT;
     }
 
-    public boolean canFailover() {
-        return Arrays.stream(NEED_FAILOVER_STATES).anyMatch(x -> x == this.getCode());
+    /**
+     * status can be take over on sub-workflow
+     * @return bool
+     */
+    public boolean canTakeover() {
+        return this == RUNNING_EXECUTION
+                || this == READY_PAUSE
+                || this == PAUSE
+                || this == READY_STOP
+                || this == STOP
+                || this == FAILURE
+                || this == SUCCESS
+                || this == FAILOVER;
     }
 
     public boolean canDirectPauseInDB() {

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkflowInstanceMapper.java

@@ -56,16 +56,6 @@ public interface WorkflowInstanceMapper extends BaseMapper<WorkflowInstance> {
     List<WorkflowInstance> queryByHostAndStatus(@Param("host") String host,
                                                 @Param("states") int[] stateArray);
 
-    /**
-     * query workflow instance by host and stateArray which is not sub workflow
-     *
-     * @param host       host
-     * @param stateArray stateArray
-     * @return workflow instance list
-     */
-    List<WorkflowInstance> queryMainWorkflowByHostAndStatus(@Param("host") String host,
-                                                            @Param("states") int[] stateArray);
-
     /**
      * query workflow instance host by stateArray
      *

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowInstanceDaoImpl.java

@@ -176,7 +176,7 @@ public List<String> queryNeedFailoverMasters() {
 
     @Override
     public List<WorkflowInstance> queryNeedFailoverWorkflowInstances(String masterAddress) {
-        return mybatisMapper.queryMainWorkflowByHostAndStatus(masterAddress,
+        return mybatisMapper.queryByHostAndStatus(masterAddress,
                 WorkflowExecutionStatus.getNeedFailoverWorkflowInstanceState());
     }
 }

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkflowInstanceMapper.xml

@@ -50,22 +50,6 @@
         </if>
         order by id asc
     </select>
-    <select id="queryMainWorkflowByHostAndStatus" resultType="org.apache.dolphinscheduler.dao.entity.WorkflowInstance">
-        select
-        <include refid="baseSql"/>
-        from t_ds_workflow_instance
-        where is_sub_workflow=0
-        <if test="host != null and host != ''">
-            and host=#{host}
-        </if>
-        <if test="states != null and states.length != 0">
-            and state in
-            <foreach collection="states" item="i" open="(" close=")" separator=",">
-                #{i}
-            </foreach>
-        </if>
-        order by id asc
-    </select>
     <select id="queryNeedFailoverWorkflowInstanceHost" resultType="String">
         select distinct host
         from t_ds_workflow_instance

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java

@@ -40,7 +40,6 @@
 import org.apache.dolphinscheduler.server.master.engine.executor.plugin.ITaskParameterDeserializer;
 import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteException;
-import org.apache.dolphinscheduler.server.master.failover.WorkflowFailover;
 import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
 import org.apache.dolphinscheduler.task.executor.events.TaskExecutorRuntimeContextChangedLifecycleEvent;
 
@@ -168,9 +167,9 @@ private SubWorkflowLogicTaskRuntimeContext recoverFromFaultTolerantTasks() {
         final WorkflowInstance subWorkflowInstance = workflowInstanceDao.queryById(
                 subWorkflowLogicTaskRuntimeContext.getSubWorkflowInstanceId());
 
-        if (subWorkflowInstance != null && subWorkflowInstance.getState().canFailover()) {
-            // Only handle sub-workflow's fail-over in SubWorkflowLogicTask's fail-over
-            applicationContext.getBean(WorkflowFailover.class).failoverWorkflow(subWorkflowInstance);
+        if (subWorkflowInstance != null && subWorkflowInstance.getState().canTakeover()) {
+            // Here we only need to take over the runtime context of sub-workflow,
+            // the sub-workflow will be failover by master-server when needed.
             return subWorkflowLogicTaskRuntimeContext;
         }