From 3eeea22671726d6797ef70e72b41db72fefd314b Mon Sep 17 00:00:00 2001 From: Zhou Zheng <1606079777@qq.comom> Date: Sat, 29 Aug 2020 20:58:31 +0800 Subject: [PATCH] improve test of DAGhelper --- .../api/service/ExecutorService2Test.java | 2 +- .../dolphinscheduler/dao/utils/DagHelper.java | 25 ++++++++----- .../dao/utils/DagHelperTest.java | 36 +++++++++++++++++++ 3 files changed, 54 insertions(+), 9 deletions(-) diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java index 0a8a99c21..8068974e5 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java @@ -265,7 +265,7 @@ public class ExecutorService2Test { .thenReturn(true); Map result = executorService.execute(loginUser, projectName, processInstanceId, ExecuteType.RESUME_FROM_FORCED_SUCCESS); - Assert.assertEquals(result.get(Constants.STATUS), Status.SUCCESS); + Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS)); verify(processService, times(1)).createCommand(any(Command.class)); } diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java index 3ad64d010..e5c4b8194 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java @@ -32,6 +32,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -278,13 +279,7 @@ public class DagHelper { if (StringUtils.isNotEmpty(parentNodeName)) { TaskNode task = dag.getNode(parentNodeName); if (task.isConditionsTask() && completeTaskList.containsKey(parentNodeName)) { - ConditionsParameters conditionsParameters = JSONUtils.parseObject(task.getConditionResult(), ConditionsParameters.class); - TaskInstance taskInstance = completeTaskList.get(parentNodeName); - if (taskInstance.getState().typeIsSuccess()) { - startVertexs = conditionsParameters.getSuccessNode(); - } else if (taskInstance.getState().typeIsFailure()) { - startVertexs = conditionsParameters.getFailedNode(); - } + startVertexs = parseConditionTask(parentNodeName, task, completeTaskList); } else { startVertexs = dag.getSubsequentNodes(parentNodeName); @@ -293,7 +288,7 @@ public class DagHelper { startVertexs = dag.getBeginNode(); } - List tmpStartVertexs = new ArrayList<>(); + Set tmpStartVertexs = new HashSet<>(); if (startVertexs != null) { tmpStartVertexs.addAll(startVertexs); } @@ -317,6 +312,20 @@ public class DagHelper { return tmpStartVertexs; } + /** + * parse condition post nodes + */ + private static Collection parseConditionTask(String parentNodeName, TaskNode task, Map completeTaskList) { + ConditionsParameters conditionsParameters = JSONUtils.parseObject(task.getConditionResult(), ConditionsParameters.class); + TaskInstance taskInstance = completeTaskList.get(parentNodeName); + if (taskInstance.getState().typeIsSuccess()) { + return conditionsParameters.getSuccessNode(); + } else if (taskInstance.getState().typeIsFailure()) { + return conditionsParameters.getFailedNode(); + } + return null; + } + /** * the task can be submit when all the depends nodes are forbidden or complete * @param taskNode taskNode diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/utils/DagHelperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/utils/DagHelperTest.java index 14dfe0b75..9e569df61 100644 --- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/utils/DagHelperTest.java +++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/utils/DagHelperTest.java @@ -19,7 +19,9 @@ package org.apache.dolphinscheduler.dao.utils; import com.fasterxml.jackson.core.JsonProcessingException; import org.apache.dolphinscheduler.common.Constants; +import org.apache.dolphinscheduler.common.enums.ExecutionStatus; import org.apache.dolphinscheduler.common.enums.TaskDependType; +import org.apache.dolphinscheduler.common.enums.TaskType; import org.apache.dolphinscheduler.common.graph.DAG; import org.apache.dolphinscheduler.common.model.TaskNode; import org.apache.dolphinscheduler.common.model.TaskNodeRelation; @@ -34,6 +36,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; /** * dag helper test @@ -79,6 +82,7 @@ public class DagHelperTest { /** * 1->2->3->5 * 4->3 + * 3->6 * @return dag * @throws JsonProcessingException if error throws JsonProcessingException */ @@ -120,6 +124,14 @@ public class DagHelperTest { node5.setDepList(dep5); taskNodeList.add(node5); + TaskNode node6 = new TaskNode(); + node6.setId("6"); + node6.setName("6"); + List dep6 = new ArrayList<>(); + dep6.add("3"); + node6.setDepList(dep6); + taskNodeList.add(node6); + List startNodes = new ArrayList<>(); List recoveryNodes = new ArrayList<>(); List destTaskNodeList = DagHelper.generateFlowNodeListByStartNode(taskNodeList, @@ -148,4 +160,28 @@ public class DagHelperTest { Assert.assertNotNull(dag); } + @Test + public void testGetStartVertex() throws JsonProcessingException { + // let 1->2->3->5, 3->5, 4->3 + DAG dag = generateDag(); + + // test when completeList is null + Assert.assertEquals(2, DagHelper.getStartVertex(null, dag, null).size()); + Assert.assertEquals(1, DagHelper.getStartVertex("1", dag, null).size()); + + // test when 3 is CONDITIONS and 1,2,3,4 all completed + Map completeTaskList = new ConcurrentHashMap<>(); + completeTaskList.putIfAbsent("1", new TaskInstance()); + completeTaskList.putIfAbsent("2", new TaskInstance()); + completeTaskList.putIfAbsent("4", new TaskInstance()); + TaskInstance task3 = new TaskInstance(); + task3.setState(ExecutionStatus.SUCCESS); + completeTaskList.putIfAbsent("3", task3); + + dag.getNode("3").setConditionResult("{\"successNode\":[\"5\"],\"failedNode\":[\"6\"]}"); + dag.getNode("3").setType(TaskType.CONDITIONS.toString()); + + Assert.assertEquals(1, DagHelper.getStartVertex(null, dag, completeTaskList).size()); + Assert.assertEquals(true, DagHelper.getStartVertex(null, dag, completeTaskList).contains("5")); + } }