test/DolphinScheduler
1
0
Fork 0
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git synced 2024-11-29 18:58:05 +08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

make sure all failed task will save in errorTaskMap (#12424)

Browse Source
This commit is contained in:
Yann Ann 2022-10-19 09:36:47 +08:00 committed by GitHub
parent ba538067f2
commit 38b643f69b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 8 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
View File

@ -391,13 +391,13 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
retryTaskInstance(taskInstance);
} else if (taskInstance.getState().isFailure()) {
completeTaskMap.put(taskInstance.getTaskCode(), taskInstance.getId());
errorTaskMap.put(taskInstance.getTaskCode(), taskInstance.getId());
// There are child nodes and the failure policy is: CONTINUE
if (processInstance.getFailureStrategy() == FailureStrategy.CONTINUE && DagHelper.haveAllNodeAfterNode(
Long.toString(taskInstance.getTaskCode()),
dag)) {
submitPostNode(Long.toString(taskInstance.getTaskCode()));
} else {
errorTaskMap.put(taskInstance.getTaskCode(), taskInstance.getId());
if (processInstance.getFailureStrategy() == FailureStrategy.END) {
killAllTasks();
}
@ -421,7 +421,6 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
/**
* release task group
*
* @param taskInstance
*/
public void releaseTaskGroup(TaskInstance taskInstance) {
logger.info("Release task group");
@ -448,7 +447,6 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
/**
* crate new task instance to retry, different objects from the original
*
* @param taskInstance
*/
private void retryTaskInstance(TaskInstance taskInstance) throws StateEventHandleException {
if (!taskInstance.taskCanRetry()) {
@ -662,10 +660,7 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
}
private boolean needComplementProcess() {
if (processInstance.isComplementData() && Flag.NO == processInstance.getIsSubProcess()) {
return true;
}
return false;
return processInstance.isComplementData() && Flag.NO == processInstance.getIsSubProcess();
}
/**
@ -1069,7 +1064,7 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
/**
* clone a new taskInstance for retry and reset some logic fields
*
* @return
* @return taskInstance
*/
public TaskInstance cloneRetryTaskInstance(TaskInstance taskInstance) {
TaskNode taskNode = dag.getNode(Long.toString(taskInstance.getTaskCode()));
@ -1097,7 +1092,7 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
/**
* clone a new taskInstance for tolerant and reset some logic fields
*
* @return
* @return taskInstance
*/
public TaskInstance cloneTolerantTaskInstance(TaskInstance taskInstance) {
TaskNode taskNode = dag.getNode(Long.toString(taskInstance.getTaskCode()));
@ -1119,9 +1114,9 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
/**
* new a taskInstance
*
* @param processInstance
* @param taskNode
* @return
* @param processInstance process instance
* @param taskNode task node
* @return task instance
*/
public TaskInstance newTaskInstance(ProcessInstance processInstance, TaskNode taskNode) {
TaskInstance taskInstance = new TaskInstance();
@ -1445,9 +1440,7 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
long taskCode = Long.parseLong(dependNodeName);
Integer taskInstanceId = completeTaskMap.get(taskCode);
TaskExecutionStatus depTaskState = taskInstanceMap.get(taskInstanceId).getState();
if (depTaskState.isFailure()) {
return false;
}
return !depTaskState.isFailure();
}
return true;
}