natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix 13347 (#13426) 

Co-authored-by: JinyLeeChina <jiny.li@foxmail.com>
This commit is contained in:
JinYong Li 2023-01-19 15:43:58 +08:00 committed by GitHub
parent 11466c934f
commit 7077d0e1ba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 14 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java
View File

@ -27,6 +27,7 @@ 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.enums.Flag;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.*;
@ -177,13 +178,21 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst
}
ProcessInstance processInstance = processService.findProcessInstanceDetailById(task.getProcessInstanceId());
if (processInstance != null && processInstance.getState().typeIsFailure()) {
if (processInstance != null && (processInstance.getState().typeIsFailure() || processInstance.getState().typeIsCancel())) {
List<TaskInstance> validTaskList = processService.findValidTaskListByProcessId(processInstance.getId());
List<Integer> failTaskList = validTaskList.stream().filter(instance -> instance.getState().typeIsFailure())
List<Long> instanceTaskCodeList = validTaskList.stream().map(TaskInstance::getTaskCode).collect(Collectors.toList());
List<ProcessTaskRelation> taskRelations = processService.findRelationByCode(processInstance.getProcessDefinitionCode(),
processInstance.getProcessDefinitionVersion());
List<TaskDefinitionLog> taskDefinitionLogs = processService.genTaskDefineList(taskRelations);
List<Long> definiteTaskCodeList = taskDefinitionLogs.stream().filter(definitionLog -> definitionLog.getFlag() == Flag.YES)
.map(TaskDefinitionLog::getCode).collect(Collectors.toList());
if (CollectionUtils.equalLists(instanceTaskCodeList, definiteTaskCodeList)) {
List<Integer> failTaskList = validTaskList.stream().filter(instance -> instance.getState().typeIsFailure() || instance.getState().typeIsCancel())
.map(TaskInstance::getId).collect(Collectors.toList());
if (failTaskList.size() == 1 && failTaskList.contains(taskInstanceId)) {
processInstance.setState(ExecutionStatus.SUCCESS);
processService.updateProcessInstance(processInstance);
if (failTaskList.size() == 1 && failTaskList.contains(taskInstanceId)) {
processInstance.setState(ExecutionStatus.SUCCESS);
processService.updateProcessInstance(processInstance);
}
}
}
// change the state of the task instance