[Feature-3575][*] Force Task Success

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskInstanceController.java

@@ -17,6 +17,7 @@
 
 package org.apache.dolphinscheduler.api.controller;
 
+import static org.apache.dolphinscheduler.api.enums.Status.FORCE_TASK_SUCCESS_ERROR;
 import static org.apache.dolphinscheduler.api.enums.Status.QUERY_TASK_LIST_PAGING_ERROR;
 
 import org.apache.dolphinscheduler.api.exceptions.ApiException;
@@ -36,6 +37,7 @@
 import org.springframework.http.HttpStatus;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestAttribute;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
@@ -127,4 +129,27 @@ public Result queryTaskListPaging(@ApiIgnore @RequestAttribute(value = Constants
         return returnDataListPaging(result);
     }
 
+    /**
+     * change one task instance's state from FAILURE to FORCED_SUCCESS
+     *
+     * @param loginUser      login user
+     * @param projectName    project name
+     * @param taskInstanceId task instance id
+     * @return the result code and msg
+     */
+    @ApiOperation(value = "force-success", notes = "FORCE_TASK_SUCCESS")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "taskInstanceId", value = "TASK_INSTANCE_ID", required = true, dataType = "Int", example = "12")
+    })
+    @PostMapping(value = "/force-success")
+    @ResponseStatus(HttpStatus.OK)
+    @ApiException(FORCE_TASK_SUCCESS_ERROR)
+    public Result<Object> forceTaskSuccess(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+                                           @ApiParam(name = "projectName", value = "PROJECT_NAME", required = true) @PathVariable String projectName,
+                                           @RequestParam(value = "taskInstanceId") Integer taskInstanceId) {
+        logger.info("force task success, project: {}, task instance id: {}", projectName, taskInstanceId);
+        Map<String, Object> result = taskInstanceService.forceTaskSuccess(loginUser, projectName, taskInstanceId);
+        return returnDataList(result);
+    }
+
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/TaskCountDto.java

@@ -57,6 +57,16 @@ private void countTaskDtos(List<ExecuteStatusCount> taskInstanceStateCounts) {
                 .sum();
     }
 
+    // remove the specified state
+    public void removeStateFromCountList(ExecutionStatus status) {
+        for (TaskStateCount count : this.taskCountDtos) {
+            if (count.getTaskStateType().equals(status)) {
+                this.taskCountDtos.remove(count);
+                break;
+            }
+        }
+    }
+
     public List<TaskStateCount> getTaskCountDtos() {
         return taskCountDtos;
     }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/ExecuteType.java

@@ -29,8 +29,9 @@ public enum ExecuteType {
      * 3 resume failure
      * 4 stop
      * 5 pause
+     * 6 resume from forced success
      */
-    NONE,REPEAT_RUNNING, RECOVER_SUSPENDED_PROCESS, START_FAILURE_TASK_PROCESS, STOP, PAUSE;
+    NONE, REPEAT_RUNNING, RECOVER_SUSPENDED_PROCESS, START_FAILURE_TASK_PROCESS, STOP, PAUSE, RESUME_FROM_FORCED_SUCCESS;
 
 
     public static ExecuteType getEnum(int value){

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java

@@ -196,7 +196,9 @@ public enum Status {
     QUERY_AUTHORIZED_AND_USER_CREATED_PROJECT_ERROR(10162, "query authorized and user created project error error", "查询授权的和用户创建的项目错误"),
     DELETE_PROCESS_DEFINITION_BY_ID_FAIL(10163,"delete process definition by id fail, for there are {0} process instances in executing using it", "删除工作流定义失败，有[{0}]个运行中的工作流实例正在使用"),
     CHECK_TENANT_CODE_ERROR(10164, "Please enter the English tenant code", "请输入英文租户编码"),
-
+    FORCE_TASK_SUCCESS_ERROR(10165, "force task success error", "强制成功任务实例错误"),
+    TASK_INSTANCE_STATE_OPERATION_ERROR(10166, "the status of task instance {0} is {1},Cannot perform force success operation", "任务实例[{0}]的状态是[{1}]，无法执行强制成功操作"),
+
     UDF_FUNCTION_NOT_EXIST(20001, "UDF function not found", "UDF函数不存在"),
     UDF_FUNCTION_EXISTS(20002, "UDF function already exists", "UDF函数已存在"),
     RESOURCE_NOT_EXIST(20004, "resource not exist", "资源不存在"),
@@ -247,8 +249,9 @@ public enum Status {
     BATCH_DELETE_PROCESS_DEFINE_BY_IDS_ERROR(50026, "batch delete process definition by ids {0} error", "批量删除工作流定义[{0}]错误"),
     TENANT_NOT_SUITABLE(50027, "there is not any tenant suitable, please choose a tenant available.", "没有合适的租户，请选择可用的租户"),
     EXPORT_PROCESS_DEFINE_BY_ID_ERROR(50028, "export process definition by id error", "导出工作流定义错误"),
-    BATCH_EXPORT_PROCESS_DEFINE_BY_IDS_ERROR(50028, "batch export process definition by ids error", "批量导出工作流定义错误"),
+    BATCH_EXPORT_PROCESS_DEFINE_BY_IDS_ERROR(50028,"batch export process definition by ids error", "批量导出工作流定义错误"),
     IMPORT_PROCESS_DEFINE_ERROR(50029, "import process definition error", "导入工作流定义错误"),
+    NO_VALID_FORCED_SUCCESS_TASK(50030, "there is no valid forced success node in process instance {0}", "工作流实例[{0}]中不包含有效的强制成功的任务实例"),
 
     HDFS_NOT_STARTUP(60001, "hdfs not startup", "hdfs未启用"),
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java

@@ -33,6 +33,7 @@
 import org.apache.dolphinscheduler.common.enums.ReleaseState;
 import org.apache.dolphinscheduler.common.enums.RunMode;
 import org.apache.dolphinscheduler.common.enums.TaskDependType;
+import org.apache.dolphinscheduler.common.enums.TaskType;
 import org.apache.dolphinscheduler.common.enums.WarningType;
 import org.apache.dolphinscheduler.common.model.Server;
 import org.apache.dolphinscheduler.common.utils.CollectionUtils;
@@ -283,6 +284,13 @@ public Map<String, Object> execute(User loginUser, String projectName, Integer p
                     result = updateProcessInstancePrepare(processInstance, CommandType.PAUSE, ExecutionStatus.READY_PAUSE);
                 }
                 break;
+            case RESUME_FROM_FORCED_SUCCESS:
+                if (!this.checkValidForcedSuccessTask(processInstanceId)) {
+                    putMsg(result, Status.NO_VALID_FORCED_SUCCESS_TASK, processInstance.getName());
+                } else {
+                    result = insertCommand(loginUser, processInstanceId, processDefinition.getId(), CommandType.RESUME_FROM_FORCED_SUCCESS);
+                }
+                break;
             default:
                 logger.error("unknown execute type : {}", executeType);
                 putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, "unknown execute type");
@@ -330,6 +338,7 @@ private Map<String, Object> checkExecuteType(ProcessInstance processInstance, Ex
                 }
                 break;
             case START_FAILURE_TASK_PROCESS:
+            case RESUME_FROM_FORCED_SUCCESS:
                 if (executionStatus.typeIsFailure()) {
                     checkResult = true;
                 }
@@ -610,4 +619,26 @@ private Map<String, Object> checkResultAndAuth(User loginUser, String projectNam
         return null;
     }
 
+    /**
+     * check if the process instance contains valid forced success task
+     *
+     * @param processInstanceId
+     * @return
+     */
+    private boolean checkValidForcedSuccessTask(int processInstanceId) {
+        List<Integer> forcedSuccessList = processService.findTaskIdByInstanceState(processInstanceId, ExecutionStatus.FORCED_SUCCESS);
+        if (forcedSuccessList != null && !forcedSuccessList.isEmpty()) {
+            return true;
+        }
+
+        List<Integer> failedSubList = processService.findTaskIdByInstanceStatusAndType(processInstanceId,
+                new ExecutionStatus[]{ExecutionStatus.FAILURE, ExecutionStatus.KILL, ExecutionStatus.NEED_FAULT_TOLERANCE},
+                TaskType.SUB_PROCESS);
+        for (int i = 0; i < failedSubList.size(); i++) {
+            if (processService.haveForcedSuccessInSubProcess(failedSubList.get(i))) {
+                return true;
+            }
+        }
+        return false;
+    }
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskInstanceService.java

@@ -145,6 +145,50 @@ public Map<String, Object> queryTaskListPaging(User loginUser, String projectNam
         return result;
     }
 
+    /**
+     * change one task instance's state from failure to forced success
+     *
+     * @param loginUser      login user
+     * @param projectName    project name
+     * @param taskInstanceId task instance id
+     * @return the result code and msg
+     */
+    public Map<String, Object> forceTaskSuccess(User loginUser, String projectName, Integer taskInstanceId) {
+        Map<String, Object> result = new HashMap<>(5);
+        Project project = projectMapper.queryByName(projectName);
+
+        // check user auth
+        Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+        Status status = (Status) checkResult.get(Constants.STATUS);
+        if (status != Status.SUCCESS) {
+            return checkResult;
+        }
+
+        // check whether the task instance can be found
+        TaskInstance task = taskInstanceMapper.selectById(taskInstanceId);
+        if (task == null) {
+            putMsg(result, Status.TASK_INSTANCE_NOT_FOUND);
+            return result;
+        }
+
+        // check whether the task instance state type is failure
+        if (!task.getState().typeIsFailure()) {
+            putMsg(result, Status.TASK_INSTANCE_STATE_OPERATION_ERROR, taskInstanceId, task.getState().toString());
+            return result;
+        }
+
+        // change the state of the task instance
+        task.setState(ExecutionStatus.FORCED_SUCCESS);
+        int changedNum = taskInstanceMapper.updateById(task);
+        if (changedNum > 0) {
+            putMsg(result, Status.SUCCESS);
+        } else {
+            putMsg(result, Status.FORCE_TASK_SUCCESS_ERROR);
+        }
+
+        return result;
+    }
+
     /***
      * generate {@link org.apache.dolphinscheduler.api.enums.Status#REQUEST_PARAMS_NOT_VALID_ERROR} res with  param name
      * @param result exist result map

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/DataAnalysisServiceImpl.java

@@ -25,6 +25,7 @@
 import org.apache.dolphinscheduler.api.service.ProjectService;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
 import org.apache.dolphinscheduler.common.enums.UserType;
 import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
@@ -114,12 +115,17 @@ public Map<String, Object> countTaskStateByProject(User loginUser, int projectId
      * @return process instance state count data
      */
     public Map<String, Object> countProcessInstanceStateByProject(User loginUser, int projectId, String startDate, String endDate) {
-        return this.countStateByProject(
+        Map<String, Object> result =  this.countStateByProject(
                 loginUser,
                 projectId,
                 startDate,
                 endDate,
                 (start, end, projectIds) -> this.processInstanceMapper.countInstanceStateByUser(start, end, projectIds));
+        // process state count needs to remove state of forced success
+        if (result.containsKey(Constants.STATUS) && result.get(Constants.STATUS).equals(Status.SUCCESS)) {
+            ((TaskCountDto)result.get(Constants.DATA_LIST)).removeStateFromCountList(ExecutionStatus.FORCED_SUCCESS);
+        }
+        return result;
     }
 
     private Map<String, Object> countStateByProject(User loginUser, int projectId, String startDate, String endDate

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/TaskInstanceControllerTest.java

@@ -18,33 +18,42 @@
 package org.apache.dolphinscheduler.api.controller;
 
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.service.TaskInstanceService;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
 import org.apache.dolphinscheduler.api.utils.Result;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.dao.entity.User;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
-import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.Mockito;
-import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MvcResult;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
 
 /**
  * task instance controller test
  */
-@RunWith(MockitoJUnitRunner.Silent.class)
-public class TaskInstanceControllerTest {
+public class TaskInstanceControllerTest extends AbstractControllerTest {
 
     @InjectMocks
     private TaskInstanceController taskInstanceController;
@@ -67,7 +76,27 @@ public void testQueryTaskListPaging() {
         Result taskResult = taskInstanceController.queryTaskListPaging(null, "", 1, "", "",
                 "", "", ExecutionStatus.SUCCESS,"192.168.xx.xx", "2020-01-01 00:00:00", "2020-01-02 00:00:00",pageNo, pageSize);
         Assert.assertEquals(Integer.valueOf(Status.SUCCESS.getCode()), taskResult.getCode());
-
     }
 
+    @Ignore
+    @Test
+    public void testForceTaskSuccess() throws Exception {
+        MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
+        paramsMap.add("taskInstanceId", "104");
+
+        Map<String, Object> mockResult = new HashMap<>(5);
+        mockResult.put(Constants.STATUS, Status.SUCCESS);
+        mockResult.put(Constants.MSG, Status.SUCCESS.getMsg());
+        when(taskInstanceService.forceTaskSuccess(any(User.class), anyString(), anyInt())).thenReturn(mockResult);
+
+        MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/task-instance/force-success", "cxc_1113")
+                .header(SESSION_ID, sessionId)
+                .params(paramsMap))
+                .andExpect(status().isOk())
+                .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
+                .andReturn();
+
+        Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
+        Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
+    }
 }