[implement] change task state
This commit is contained in:
Zhou Zheng
parent
7f8d060dca
commit
d86ceeecbe
|
|
@ -34,6 +34,7 @@ import springfox.documentation.annotations.ApiIgnore;
|
|||
|
||||
import java.util.Map;
|
||||
|
||||
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;
|
||||
|
||||
/**
|
||||
|
|
@ -103,4 +104,28 @@ public class TaskInstanceController extends BaseController {
|
|||
return returnDataListPaging(result);
|
||||
}
|
||||
|
||||
/**
|
||||
* change one single 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 = "")
|
||||
@ApiImplicitParams({
|
||||
@ApiImplicitParam(name = "taskInstanceId", value = "TASK_INTSTANCE_ID", required = true, dataType = "Int", example = "2")
|
||||
})
|
||||
@PostMapping(value = "/force-success")
|
||||
@ResponseStatus(HttpStatus.OK)
|
||||
@ApiException(FORCE_TASK_SUCCESS_ERROR)
|
||||
public Result forceSingleTaskSuccess(@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, login user: {}, project:{}, task instance id:{}",
|
||||
loginUser.getUserName(), projectName, taskInstanceId);
|
||||
Map<String, Object> result = taskInstanceService.forceSingleTaskSuccess(loginUser, projectName, taskInstanceId);
|
||||
return returnDataList(result);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -176,6 +176,8 @@ public enum Status {
|
|||
DELETE_WORKER_GROUP_FAIL(10147,"delete worker group fail ", "删除worker分组失败"),
|
||||
COPY_PROCESS_DEFINITION_ERROR(10148,"copy process definition error", "复制工作流错误"),
|
||||
USER_DISABLED(10149,"The current user is disabled", "当前用户已停用"),
|
||||
FORCE_TASK_SUCCESS_ERROR(10150, "force task success error", "强制成功任务实例错误"),
|
||||
TASK_INSTANCE_STATE_OPETATION_ERROR(10151, "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函数已存在"),
|
||||
|
|
|
|||
|
|
@ -138,4 +138,51 @@ public class TaskInstanceService extends BaseService {
|
|||
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* change one single 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> forceSingleTaskSuccess(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()) {
|
||||
// FIXME: 这个status的msg中是包含参数的，需要处理一下（已完善，待测试）
|
||||
putMsg(result, Status.TASK_INSTANCE_STATE_OPETATION_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 {
|
||||
// FIXME: 或许应该再加一个状态码
|
||||
putMsg(result, Status.FORCE_TASK_SUCCESS_ERROR);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -62,4 +62,9 @@ public class TaskInstanceControllerTest extends AbstractControllerTest{
|
|||
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
|
||||
logger.info(mvcResult.getResponse().getContentAsString());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void forceSingleTaskSuccess() {
|
||||
// TODO： 加入测试
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -189,4 +189,9 @@ public class TaskInstanceServiceTest {
|
|||
result.put(Constants.MSG, status.getMsg());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void forceSingleTaskSuccess() {
|
||||
// TODO: 加入测试
|
||||
}
|
||||
}
|
||||
|
|
@ -41,6 +41,7 @@ public enum ExecutionStatus {
|
|||
* 9 kill
|
||||
* 10 waiting thread
|
||||
* 11 waiting depend node complete
|
||||
* 12 forced success
|
||||
*/
|
||||
SUBMITTED_SUCCESS(0, "submit success"),
|
||||
RUNNING_EXEUTION(1, "running"),
|
||||
|
|
@ -53,7 +54,8 @@ public enum ExecutionStatus {
|
|||
NEED_FAULT_TOLERANCE(8, "need fault tolerance"),
|
||||
KILL(9, "kill"),
|
||||
WAITTING_THREAD(10, "waiting thread"),
|
||||
WAITTING_DEPEND(11, "waiting depend node complete");
|
||||
WAITTING_DEPEND(11, "waiting depend node complete"),
|
||||
FORCED_SUCCESS(12, "forced success");
|
||||
|
||||
ExecutionStatus(int code, String descp){
|
||||
this.code = code;
|
||||
|
|
@ -77,7 +79,7 @@ public enum ExecutionStatus {
|
|||
* @return status
|
||||
*/
|
||||
public boolean typeIsSuccess(){
|
||||
return this == SUCCESS;
|
||||
return this == SUCCESS || this == FORCED_SUCCESS;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Reference in New Issue