Skip to content

Commit

Permalink
解决单独执行子节点空指针的问题(增加checkstyle)
Browse files Browse the repository at this point in the history
  • Loading branch information
liang.li.c committed Nov 23, 2020
1 parent 7897214 commit 86567e0
Showing 1 changed file with 31 additions and 16 deletions.
47 changes: 31 additions & 16 deletions dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,10 +70,10 @@ public static List<TaskNodeRelation> generateRelationListByFlowNodes(List<TaskNo
/**
* generate task nodes needed by dag
*
* @param taskNodeList taskNodeList
* @param startNodeNameList startNodeNameList
* @param taskNodeList taskNodeList
* @param startNodeNameList startNodeNameList
* @param recoveryNodeNameList recoveryNodeNameList
* @param taskDependType taskDependType
* @param taskDependType taskDependType
* @return task node list
*/
public static List<TaskNode> generateFlowNodeListByStartNode(List<TaskNode> taskNodeList, List<String> startNodeNameList,
Expand Down Expand Up @@ -131,7 +131,7 @@ public static List<TaskNode> generateFlowNodeListByStartNode(List<TaskNode> task
/**
* find all the nodes that depended on the start node
*
* @param startNode startNode
* @param startNode startNode
* @param taskNodeList taskNodeList
* @return task node list
*/
Expand All @@ -156,9 +156,9 @@ private static List<TaskNode> getFlowNodeListPost(TaskNode startNode, List<TaskN
/**
* find all nodes that start nodes depend on.
*
* @param startNode startNode
* @param startNode startNode
* @param recoveryNodeNameList recoveryNodeNameList
* @param taskNodeList taskNodeList
* @param taskNodeList taskNodeList
* @return task node list
*/
private static List<TaskNode> getFlowNodeListPre(TaskNode startNode, List<String> recoveryNodeNameList, List<TaskNode> taskNodeList, List<String> visitedNodeNameList) {
Expand Down Expand Up @@ -192,9 +192,9 @@ private static List<TaskNode> getFlowNodeListPre(TaskNode startNode, List<String
* generate dag by start nodes and recovery nodes
*
* @param processDefinitionJson processDefinitionJson
* @param startNodeNameList startNodeNameList
* @param recoveryNodeNameList recoveryNodeNameList
* @param depNodeType depNodeType
* @param startNodeNameList startNodeNameList
* @param recoveryNodeNameList recoveryNodeNameList
* @param depNodeType depNodeType
* @return process dag
* @throws Exception if error throws Exception
*/
Expand Down Expand Up @@ -246,7 +246,7 @@ public static Map<String, TaskNode> getForbiddenTaskNodeMaps(String processDefin
* find node by node name
*
* @param nodeDetails nodeDetails
* @param nodeName nodeName
* @param nodeName nodeName
* @return task node
*/
public static TaskNode findNodeByName(List<TaskNode> nodeDetails, String nodeName) {
Expand All @@ -261,8 +261,8 @@ public static TaskNode findNodeByName(List<TaskNode> nodeDetails, String nodeNam
/**
* the task can be submit when all the depends nodes are forbidden or complete
*
* @param taskNode taskNode
* @param dag dag
* @param taskNode taskNode
* @param dag dag
* @param completeTaskList completeTaskList
* @return can submit
*/
Expand All @@ -276,10 +276,7 @@ public static boolean allDependsForbiddenOrEnd(TaskNode taskNode,
}
for (String dependNodeName : dependList) {
TaskNode dependNode = dag.getNode(dependNodeName);
if (dependNode == null) {
continue;
}
if (completeTaskList.containsKey(dependNodeName)
if (dependNode == null || completeTaskList.containsKey(dependNodeName)
|| dependNode.isForbidden()
|| skipTaskNodeList.containsKey(dependNodeName)) {
continue;
Expand All @@ -295,6 +292,7 @@ public static boolean allDependsForbiddenOrEnd(TaskNode taskNode,
* this function parse the condition node to find the right branch.
* also check all the depends nodes forbidden or complete
*
* @param preNodeName
* @return successor nodes
*/
public static Set<String> parsePostNodes(String preNodeName,
Expand Down Expand Up @@ -331,6 +329,9 @@ public static Set<String> parsePostNodes(String preNodeName,

/**
* if all of the task dependence are skipped, skip it too.
*
* @param taskNode
* @return
*/
private static boolean isTaskNodeNeedSkip(TaskNode taskNode,
Map<String, TaskNode> skipTaskNodeList
Expand All @@ -350,6 +351,9 @@ private static boolean isTaskNodeNeedSkip(TaskNode taskNode,
/**
* parse condition task find the branch process
* set skip flag for another one.
*
* @param nodeName
* @return
*/
public static List<String> parseConditionTask(String nodeName,
Map<String, TaskNode> skipTaskNodeList,
Expand Down Expand Up @@ -384,6 +388,11 @@ public static List<String> parseConditionTask(String nodeName,

/**
* set task node and the post nodes skip flag
*
* @param skipNodeName
* @param dag
* @param completeTaskList
* @param skipTaskNodeList
*/
private static void setTaskNodeSkip(String skipNodeName,
DAG<String, TaskNode, TaskNodeRelation> dag,
Expand Down Expand Up @@ -455,6 +464,9 @@ public static ProcessDag getProcessDag(List<TaskNode> taskNodeList) {

/**
* is there have conditions after the parent node
*
* @param parentNodeName
* @return
*/
public static boolean haveConditionsAfterNode(String parentNodeName,
DAG<String, TaskNode, TaskNodeRelation> dag
Expand All @@ -476,6 +488,9 @@ public static boolean haveConditionsAfterNode(String parentNodeName,

/**
* is there have conditions after the parent node
*
* @param parentNodeName
* @return
*/
public static boolean haveConditionsAfterNode(String parentNodeName,
List<TaskNode> taskNodes
Expand Down

0 comments on commit 86567e0

Please sign in to comment.