[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

@@ -30,7 +30,7 @@ public enum ExecuteType {
      * 4 stop
      * 5 pause
      */
-    NONE,REPEAT_RUNNING, RECOVER_SUSPENDED_PROCESS, START_FAILURE_TASK_PROCESS, STOP, PAUSE;
+    NONE, REPEAT_RUNNING, RECOVER_SUSPENDED_PROCESS, START_FAILURE_TASK_PROCESS, STOP, PAUSE;
 
 
     public static ExecuteType getEnum(int value){

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

@@ -196,6 +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_OS_TENANT_CODE_ERROR(10164, "Please enter the English os 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函数已存在"),
@@ -247,7 +250,7 @@ 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", "导入工作流定义错误"),
 
     HDFS_NOT_STARTUP(60001, "hdfs not startup", "hdfs未启用"),

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());
+    }
 }

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java

@@ -25,12 +25,14 @@
 import org.apache.dolphinscheduler.api.service.impl.ProjectServiceImpl;
 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.Priority;
 import org.apache.dolphinscheduler.common.enums.ReleaseState;
 import org.apache.dolphinscheduler.common.enums.RunMode;
 import org.apache.dolphinscheduler.common.model.Server;
 import org.apache.dolphinscheduler.dao.entity.Command;
 import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
 import org.apache.dolphinscheduler.dao.entity.Project;
 import org.apache.dolphinscheduler.dao.entity.Schedule;
 import org.apache.dolphinscheduler.dao.entity.Tenant;
@@ -82,12 +84,16 @@ public class ExecutorService2Test {
 
     private int processDefinitionId = 1;
 
+    private int processInstanceId = 1;
+
     private int tenantId = 1;
 
     private int userId = 1;
 
     private ProcessDefinition processDefinition = new ProcessDefinition();
 
+    private ProcessInstance processInstance = new ProcessInstance();
+
     private User loginUser = new User();
 
     private String projectName = "projectName";
@@ -107,6 +113,13 @@ public void init() {
         processDefinition.setTenantId(tenantId);
         processDefinition.setUserId(userId);
 
+        // processInstance
+        processInstance.setId(processInstanceId);
+        processInstance.setProcessDefinitionId(processDefinitionId);
+        processInstance.setState(ExecutionStatus.FAILURE);
+        processInstance.setExecutorId(userId);
+        processInstance.setTenantId(tenantId);
+
         // project
         project.setName(projectName);
 
@@ -120,6 +133,8 @@ public void init() {
         Mockito.when(processService.getTenantForProcess(tenantId, userId)).thenReturn(new Tenant());
         Mockito.when(processService.createCommand(any(Command.class))).thenReturn(1);
         Mockito.when(monitorService.getServerListFromZK(true)).thenReturn(getMasterServersList());
+        Mockito.when(processService.findProcessInstanceDetailById(processInstanceId)).thenReturn(processInstance);
+        Mockito.when(processService.findProcessDefineById(processDefinitionId)).thenReturn(processDefinition);
     }
 
     /**

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskInstanceServiceTest.java

@@ -214,4 +214,48 @@ private void putMsg(Map<String, Object> result, Status status, Object... statusP
             result.put(Constants.MSG, status.getMsg());
         }
     }
-}
+
+    @Test
+    public void forceTaskSuccess() {
+        User user = getAdminUser();
+        String projectName = "test";
+        Project project = getProject(projectName);
+        int taskId = 1;
+        TaskInstance task = getTaskInstance();
+
+        Map<String, Object> mockSuccess = new HashMap<>(5);
+        putMsg(mockSuccess, Status.SUCCESS);
+        when(projectMapper.queryByName(projectName)).thenReturn(project);
+
+        // user auth failed
+        Map<String, Object> mockFailure = new HashMap<>(5);
+        putMsg(mockFailure, Status.USER_NO_OPERATION_PROJECT_PERM, user.getUserName(), projectName);
+        when(projectService.checkProjectAndAuth(user, project, projectName)).thenReturn(mockFailure);
+        Map<String, Object> authFailRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
+        Assert.assertNotSame(Status.SUCCESS, authFailRes.get(Constants.STATUS));
+
+        // test task not found
+        when(projectService.checkProjectAndAuth(user, project, projectName)).thenReturn(mockSuccess);
+        when(taskInstanceMapper.selectById(Mockito.anyInt())).thenReturn(null);
+        Map<String, Object> taskNotFoundRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
+        Assert.assertEquals(Status.TASK_INSTANCE_NOT_FOUND, taskNotFoundRes.get(Constants.STATUS));
+
+        // test task instance state error
+        task.setState(ExecutionStatus.SUCCESS);
+        when(taskInstanceMapper.selectById(1)).thenReturn(task);
+        Map<String, Object> taskStateErrorRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
+        Assert.assertEquals(Status.TASK_INSTANCE_STATE_OPERATION_ERROR, taskStateErrorRes.get(Constants.STATUS));
+
+        // test error
+        task.setState(ExecutionStatus.FAILURE);
+        when(taskInstanceMapper.updateById(task)).thenReturn(0);
+        Map<String, Object> errorRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
+        Assert.assertEquals(Status.FORCE_TASK_SUCCESS_ERROR, errorRes.get(Constants.STATUS));
+
+        // test success
+        task.setState(ExecutionStatus.FAILURE);
+        when(taskInstanceMapper.updateById(task)).thenReturn(1);
+        Map<String, Object> successRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
+        Assert.assertEquals(Status.SUCCESS, successRes.get(Constants.STATUS));
+    }
+}

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

@@ -41,6 +41,7 @@ public enum ExecutionStatus {
      * 10 waiting thread
      * 11 waiting depend node complete
      * 12 delay execution
+     * 13 forced success
      */
     SUBMITTED_SUCCESS(0, "submit success"),
     RUNNING_EXECUTION(1, "running"),
@@ -54,7 +55,8 @@ public enum ExecutionStatus {
     KILL(9, "kill"),
     WAITTING_THREAD(10, "waiting thread"),
     WAITTING_DEPEND(11, "waiting depend node complete"),
-    DELAY_EXECUTION(12, "delay execution");
+    DELAY_EXECUTION(12, "delay execution"),
+    FORCED_SUCCESS(13, "forced success");
 
     ExecutionStatus(int code, String descp) {
         this.code = code;
@@ -79,7 +81,7 @@ public enum ExecutionStatus {
      * @return status
      */
     public boolean typeIsSuccess() {
-        return this == SUCCESS;
+        return this == SUCCESS || this == FORCED_SUCCESS;
     }
 
     /**

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/VarPoolUtils.java

@@ -63,7 +63,7 @@ public static void setTaskNodeLocalParams(TaskNode taskNode, Map<String, Object>
      * @throws ParseException ParseException
      */
     public static void convertVarPoolToMap(Map<String, Object> propToValue, String varPool) throws ParseException {
-        if (varPool == null || propToValue == null) {
+        if (propToValue == null || StringUtils.isEmpty(varPool)) {
             return;
         }
         String[] splits = varPool.split("\\$VarPool\\$");