[Fix-17732] Change workflow instance status to failure when command handle failed

Author: ruanwenjun

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

@@ -138,6 +138,8 @@ int updateWorkflowInstanceState(
                                     @Param("originState") WorkflowExecutionStatus originState,
                                     @Param("targetState") WorkflowExecutionStatus targetState);
 
+    int forceUpdateWorkflowInstanceState(@Param("id") Integer id, @Param("status") WorkflowExecutionStatus status);
+
     /**
      * update workflow instance by tenantCode
      *

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowInstanceDao.java

@@ -39,6 +39,8 @@ void updateWorkflowInstanceState(Integer workflowInstanceId,
                                      WorkflowExecutionStatus originState,
                                      WorkflowExecutionStatus targetState);
 
+    void forceUpdateWorkflowInstanceState(Integer id, WorkflowExecutionStatus status);
+
     /**
      * performs an "upsert" operation (update or insert) on a WorkflowInstance object within a new transaction
      *

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

@@ -75,6 +75,11 @@ public void updateWorkflowInstanceState(Integer workflowInstanceId, WorkflowExec
         }
     }
 
+    @Override
+    public void forceUpdateWorkflowInstanceState(Integer id, WorkflowExecutionStatus status) {
+        mybatisMapper.forceUpdateWorkflowInstanceState(id, status);
+    }
+
     @Override
     @Transactional(propagation = Propagation.REQUIRES_NEW, isolation = Isolation.READ_COMMITTED, rollbackFor = Exception.class)
     public void performTransactionalUpsert(WorkflowInstance workflowInstance) {

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

@@ -158,6 +158,10 @@
         where id = #{workflowInstanceId} and state = #{originState}
     </update>
 
+    <update id="forceUpdateWorkflowInstanceState">
+        update t_ds_workflow_instance set state = #{status} where id = #{id}
+    </update>
+
     <update id="updateWorkflowInstanceByTenantCode">
         update t_ds_workflow_instance
         set tenant_code = #{destTenantCode}

dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowInstanceDaoImplTest.java

@@ -96,6 +96,15 @@ void updateWorkflowInstanceState_failed() {
                 unsupportedOperationException.getMessage());
     }
 
+    @Test
+    void forceUpdateWorkflowInstanceState() {
+        WorkflowInstance workflowInstance = createWorkflowInstance(1L, 1, WorkflowExecutionStatus.RUNNING_EXECUTION);
+        workflowInstanceDao.insert(workflowInstance);
+        workflowInstanceDao.forceUpdateWorkflowInstanceState(workflowInstance.getId(), WorkflowExecutionStatus.FAILURE);
+        assertEquals(WorkflowExecutionStatus.FAILURE,
+                workflowInstanceDao.queryById(workflowInstance.getId()).getState());
+    }
+
     @Test
     void queryByWorkflowCodeVersionStatus_EXIST_FINISH_INSTANCE() {
         long workflowDefinitionCode = 1L;

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

@@ -26,6 +26,7 @@
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.dao.entity.Command;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.meter.metrics.MetricsProvider;
 import org.apache.dolphinscheduler.meter.metrics.SystemMetrics;
 import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
@@ -52,6 +53,7 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionTemplate;
 
 /**
  * Master scheduler thread, this thread will consume the commands from database and trigger processInstance executed.
@@ -75,6 +77,9 @@ public class CommandEngine extends BaseDaemonThread implements AutoCloseable {
     @Autowired
     private IWorkflowRepository workflowRepository;
 
+    @Autowired
+    private WorkflowInstanceDao workflowInstanceDao;
+
     @Autowired
     private WorkflowExecutionRunnableFactory workflowExecutionRunnableFactory;
 
@@ -84,6 +89,9 @@ public class CommandEngine extends BaseDaemonThread implements AutoCloseable {
     @Autowired
     private WorkflowEventBusCoordinator workflowEventBusCoordinator;
 
+    @Autowired
+    private TransactionTemplate transactionTemplate;
+
     private ExecutorService commandHandleThreadPool;
 
     private boolean flag = false;
@@ -189,8 +197,18 @@ private Void bootstrapError(Command command, Throwable throwable) {
                     throwable);
             return null;
         }
-        log.error("Failed bootstrap command {} ", JSONUtils.toPrettyJsonString(command), throwable);
-        commandService.moveToErrorCommand(command, ExceptionUtils.getStackTrace(throwable));
+
+        transactionTemplate.execute(status -> {
+            log.warn("Failed bootstrap command {} ", JSONUtils.toPrettyJsonString(command), throwable);
+            final int workflowInstanceId = command.getWorkflowInstanceId();
+
+            workflowInstanceDao.forceUpdateWorkflowInstanceState(workflowInstanceId, WorkflowExecutionStatus.FAILURE);
+            log.info("Set workflow instance {} state to FAILURE", workflowInstanceId);
+
+            commandService.moveToErrorCommand(command, ExceptionUtils.getStackTrace(throwable));
+            log.info("Move command {} to error command table", command.getId());
+            return null;
+        });
         return null;
     }
 

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

@@ -28,7 +28,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public class WorkflowGraph implements IWorkflowGraph {
@@ -46,12 +45,20 @@ public WorkflowGraph(List<WorkflowTaskRelation> workflowTaskRelations, List<Task
         this.predecessors = new HashMap<>();
         this.successors = new HashMap<>();
 
-        this.taskDefinitionMap = taskDefinitions
-                .stream()
-                .collect(Collectors.toMap(TaskDefinition::getName, Function.identity()));
-        this.taskDefinitionCodeMap = taskDefinitions
-                .stream()
-                .collect(Collectors.toMap(TaskDefinition::getCode, Function.identity()));
+        this.taskDefinitionMap = new HashMap<>(taskDefinitions.size());
+        this.taskDefinitionCodeMap = new HashMap<>(taskDefinitions.size());
+        for (TaskDefinition taskDefinition : taskDefinitions) {
+            if (taskDefinitionMap.containsKey(taskDefinition.getName())) {
+                throw new IllegalArgumentException(
+                        "Duplicate task name: " + taskDefinition.getName() + " in the workflow");
+            }
+            taskDefinitionMap.put(taskDefinition.getName(), taskDefinition);
+            if (taskDefinitionCodeMap.containsKey(taskDefinition.getCode())) {
+                throw new IllegalArgumentException(
+                        "Duplicate task code: " + taskDefinition.getCode() + " in the workflow");
+            }
+            taskDefinitionCodeMap.put(taskDefinition.getCode(), taskDefinition);
+        }
 
         addTaskNodes(taskDefinitions);
         addTaskEdge(workflowTaskRelations);

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/command/CommandServiceImpl.java

@@ -21,6 +21,7 @@
 
 import org.apache.dolphinscheduler.common.constants.Constants;
 import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.dao.entity.Command;
 import org.apache.dolphinscheduler.dao.entity.ErrorCommand;
@@ -29,6 +30,7 @@
 import org.apache.dolphinscheduler.dao.mapper.ErrorCommandMapper;
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
 import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionMapper;
+import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -41,6 +43,7 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.micrometer.core.annotation.Counted;
@@ -64,11 +67,18 @@ public class CommandServiceImpl implements CommandService {
     @Autowired
     private WorkflowDefinitionMapper processDefineMapper;
 
+    @Autowired
+    private WorkflowInstanceDao workflowInstanceDao;
+
     @Override
+    @Transactional
     public void moveToErrorCommand(Command command, String message) {
-        ErrorCommand errorCommand = new ErrorCommand(command, message);
-        this.errorCommandMapper.insert(errorCommand);
-        this.commandMapper.deleteById(command.getId());
+        final ErrorCommand errorCommand = new ErrorCommand(command, message);
+        errorCommandMapper.insert(errorCommand);
+        commandMapper.deleteById(command.getId());
+        workflowInstanceDao.forceUpdateWorkflowInstanceState(command.getWorkflowInstanceId(),
+                WorkflowExecutionStatus.FAILURE);
+        log.info("Set workflow instance {} state to FAILURE", command.getWorkflowInstanceId());
     }
 
     @Override