test/DolphinScheduler
1
0
Fork 0
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git synced 2024-12-12 12:25:16 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor depend result

Browse Source
This commit is contained in:
lenboo 2020-08-02 11:13:50 +08:00
parent 0e885cd0ee
commit cdd76d6b11
1 changed files with 23 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
View File

@ -675,7 +675,6 @@ public class MasterExecThread implements Runnable {
if(!completeTaskList.containsKey(depsNode)){
return DependResult.WAITING;
}
// depend node has already complete.
ExecutionStatus depTaskState = completeTaskList.get(depsNode).getState();
if(depTaskState.typeIsPause() || depTaskState.typeIsCancel()){
return DependResult.WAITING;
@ -684,13 +683,7 @@ public class MasterExecThread implements Runnable {
if(taskNode.isConditionsTask()){
continue;
}
if(dag.getNode(depsNode).isConditionsTask()){
//condition task need check the branch to run
List<String> nextTaskList = parseConditionTask(depsNode);
if(!nextTaskList.contains(taskName)){
return DependResult.FAILED;
}
}else if(depTaskState.typeIsFailure()){
if(!dependTaskSuccess(depsNode, taskName)){
return DependResult.FAILED;
}
}
@ -698,6 +691,28 @@ public class MasterExecThread implements Runnable {
return DependResult.SUCCESS;
}
/**
* depend node is completed, but here need check the condition task branch is the next node
* @param dependNodeName
* @param nextNodeName
* @return
*/
private boolean dependTaskSuccess(String dependNodeName, String nextNodeName){
if(dag.getNode(dependNodeName).isConditionsTask()){
//condition task need check the branch to run
List<String> nextTaskList = parseConditionTask(dependNodeName);
if(!nextTaskList.contains(nextNodeName)){
return false;
}
}else {
ExecutionStatus depTaskState = completeTaskList.get(dependNodeName).getState();
if(depTaskState.typeIsFailure()){
return false;
}
}
return true;
}
/**
* query task instance by complete state