From d0e907574e5199e66b2b920523c798164e4ae9cf Mon Sep 17 00:00:00 2001 From: shangeyao Date: Thu, 25 Jun 2026 17:24:39 +0800 Subject: [PATCH 1/3] [Improvement-17057][Master] Check if the WorkflowGraph is a DAG at constructor Validate workflow task relations with Kahn topological sort when building WorkflowGraph on the master, so cyclic graphs fail fast before execution. Co-authored-by: Cursor --- .../master/engine/graph/WorkflowGraph.java | 50 ++++ .../graph/WorkflowGraphCheckIfDAGTest.java | 222 ++++++++++++++++++ 2 files changed, 272 insertions(+) create mode 100644 dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java index 5245c7c595af..067912ac4149 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java @@ -22,11 +22,13 @@ import org.apache.dolphinscheduler.dao.entity.TaskDefinition; import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Queue; import java.util.Set; import java.util.stream.Collectors; @@ -42,6 +44,7 @@ public class WorkflowGraph implements IWorkflowGraph { public WorkflowGraph(List workflowTaskRelations, List taskDefinitions) { checkNotNull(taskDefinitions, "taskDefinitions can not be null"); checkNotNull(workflowTaskRelations, "taskDefinitions can not be null"); + checkIfDAG(workflowTaskRelations, taskDefinitions); this.predecessors = new HashMap<>(); this.successors = new HashMap<>(); @@ -64,6 +67,53 @@ public WorkflowGraph(List workflowTaskRelations, List workflowTaskRelations, List taskDefinitions) { + Map> preTaskCodeMap = workflowTaskRelations + .stream() + .filter(relation -> relation.getPreTaskCode() > 0 && relation.getPostTaskCode() > 0) + .collect(Collectors.groupingBy(WorkflowTaskRelation::getPostTaskCode, + Collectors.mapping(WorkflowTaskRelation::getPreTaskCode, Collectors.toList()))); + Map> postTaskCodeMap = workflowTaskRelations + .stream() + .filter(relation -> relation.getPreTaskCode() > 0 && relation.getPostTaskCode() > 0) + .collect(Collectors.groupingBy(WorkflowTaskRelation::getPreTaskCode, + Collectors.mapping(WorkflowTaskRelation::getPostTaskCode, Collectors.toList()))); + + Map inDegreeCount = new HashMap<>(); + for (TaskDefinition taskDefinition : taskDefinitions) { + List preTasks = preTaskCodeMap.get(taskDefinition.getCode()); + inDegreeCount.put(taskDefinition.getCode(), preTasks == null ? 0 : preTasks.size()); + } + + Queue queue = new ArrayDeque<>(); + for (Map.Entry entry : inDegreeCount.entrySet()) { + if (entry.getValue() == 0) { + queue.offer(entry.getKey()); + } + } + + Set sortedTaskCodes = new HashSet<>(); + while (!queue.isEmpty()) { + Long taskCode = queue.poll(); + sortedTaskCodes.add(taskCode); + + List postCodes = postTaskCodeMap.get(taskCode); + if (postCodes == null) { + continue; + } + for (Long postCode : postCodes) { + inDegreeCount.put(postCode, inDegreeCount.get(postCode) - 1); + if (inDegreeCount.get(postCode) == 0) { + queue.offer(postCode); + } + } + } + + if (sortedTaskCodes.size() < taskDefinitions.size()) { + throw new IllegalArgumentException("The workflow task relation is not a DAG"); + } + } + @Override public List getStartNodes() { return predecessors.entrySet() diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java new file mode 100644 index 000000000000..b4005038afb0 --- /dev/null +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java @@ -0,0 +1,222 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.server.master.engine.graph; + +import org.apache.dolphinscheduler.dao.entity.TaskDefinition; +import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class WorkflowGraphCheckIfDAGTest { + + private List taskDefinitions; + private List workflowTaskRelations; + + @BeforeEach + public void setUp() { + taskDefinitions = new ArrayList<>(); + workflowTaskRelations = new ArrayList<>(); + } + + @Test + public void checkIfDAGSingleTaskNoException() { + TaskDefinition taskDefinition = new TaskDefinition(1L, 1); + taskDefinitions.add(taskDefinition); + + WorkflowTaskRelation relation = new WorkflowTaskRelation(); + relation.setPreTaskCode(0); + relation.setPostTaskCode(1); + workflowTaskRelations.add(relation); + + Assertions.assertDoesNotThrow(() -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + } + + @Test + public void checkIfDAGSimpleDAGNoException() { + TaskDefinition task1 = new TaskDefinition(1L, 1); + task1.setName("task1"); + TaskDefinition task2 = new TaskDefinition(2L, 1); + task2.setName("task2"); + taskDefinitions.add(task1); + taskDefinitions.add(task2); + + WorkflowTaskRelation relation = new WorkflowTaskRelation(); + relation.setPreTaskCode(1L); + relation.setPreTaskVersion(1); + relation.setPostTaskCode(2L); + relation.setPostTaskVersion(1); + workflowTaskRelations.add(relation); + + Assertions.assertDoesNotThrow(() -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + } + + @Test + public void checkIfDAGCycleThrowsException() { + TaskDefinition task1 = new TaskDefinition(1L, 1); + TaskDefinition task2 = new TaskDefinition(2L, 1); + taskDefinitions.add(task1); + taskDefinitions.add(task2); + + WorkflowTaskRelation relation1 = new WorkflowTaskRelation(); + relation1.setPreTaskCode(1L); + relation1.setPreTaskVersion(1); + relation1.setPostTaskCode(2L); + relation1.setPostTaskVersion(1); + WorkflowTaskRelation relation2 = new WorkflowTaskRelation(); + relation2.setPreTaskCode(2L); + relation2.setPreTaskVersion(1); + relation2.setPostTaskCode(1L); + relation2.setPostTaskVersion(1); + workflowTaskRelations.add(relation1); + workflowTaskRelations.add(relation2); + + Assertions.assertThrows(IllegalArgumentException.class, + () -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + } + + @Test + public void checkIfDAGMultipleZeroInDegreeNoException() { + TaskDefinition task1 = new TaskDefinition(1L, 1); + task1.setName("task1"); + TaskDefinition task2 = new TaskDefinition(2L, 1); + task2.setName("task2"); + TaskDefinition task3 = new TaskDefinition(3L, 1); + task3.setName("task3"); + taskDefinitions.add(task1); + taskDefinitions.add(task2); + taskDefinitions.add(task3); + + WorkflowTaskRelation relation1 = new WorkflowTaskRelation(); + relation1.setPreTaskCode(1L); + relation1.setPreTaskVersion(1); + relation1.setPostTaskCode(2L); + relation1.setPostTaskVersion(1); + WorkflowTaskRelation relation2 = new WorkflowTaskRelation(); + relation2.setPreTaskCode(1L); + relation2.setPreTaskVersion(1); + relation2.setPostTaskCode(3L); + relation2.setPostTaskVersion(1); + workflowTaskRelations.add(relation1); + workflowTaskRelations.add(relation2); + + Assertions.assertDoesNotThrow(() -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + } + + @Test + public void checkIfDAGComplexDAGNoException() { + TaskDefinition task1 = new TaskDefinition(1L, 1); + task1.setName("task1"); + TaskDefinition task2 = new TaskDefinition(2L, 1); + task2.setName("task2"); + TaskDefinition task3 = new TaskDefinition(3L, 1); + task3.setName("task3"); + TaskDefinition task4 = new TaskDefinition(4L, 1); + task4.setName("task4"); + taskDefinitions.add(task1); + taskDefinitions.add(task2); + taskDefinitions.add(task3); + taskDefinitions.add(task4); + + WorkflowTaskRelation relation1 = new WorkflowTaskRelation(); + relation1.setPreTaskCode(1L); + relation1.setPreTaskVersion(1); + relation1.setPostTaskCode(2L); + relation1.setPostTaskVersion(1); + + WorkflowTaskRelation relation2 = new WorkflowTaskRelation(); + relation2.setPreTaskCode(1L); + relation2.setPreTaskVersion(1); + relation2.setPostTaskCode(3L); + relation2.setPostTaskVersion(1); + + WorkflowTaskRelation relation3 = new WorkflowTaskRelation(); + relation3.setPreTaskCode(2L); + relation3.setPreTaskVersion(1); + relation3.setPostTaskCode(4L); + relation3.setPostTaskVersion(1); + + WorkflowTaskRelation relation4 = new WorkflowTaskRelation(); + relation4.setPreTaskCode(3L); + relation4.setPreTaskVersion(1); + relation4.setPostTaskCode(4L); + relation4.setPostTaskVersion(1); + + workflowTaskRelations.add(relation1); + workflowTaskRelations.add(relation2); + workflowTaskRelations.add(relation3); + workflowTaskRelations.add(relation4); + + Assertions.assertDoesNotThrow(() -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + } + + @Test + public void checkIfDAGComplexDAGThrowsException() { + TaskDefinition task1 = new TaskDefinition(1L, 1); + TaskDefinition task2 = new TaskDefinition(2L, 1); + TaskDefinition task3 = new TaskDefinition(3L, 1); + TaskDefinition task4 = new TaskDefinition(4L, 1); + taskDefinitions.add(task1); + taskDefinitions.add(task2); + taskDefinitions.add(task3); + taskDefinitions.add(task4); + + WorkflowTaskRelation relation1 = new WorkflowTaskRelation(); + relation1.setPreTaskCode(1L); + relation1.setPreTaskVersion(1); + relation1.setPostTaskCode(2L); + relation1.setPostTaskVersion(1); + + WorkflowTaskRelation relation2 = new WorkflowTaskRelation(); + relation2.setPreTaskCode(1L); + relation2.setPreTaskVersion(1); + relation2.setPostTaskCode(3L); + relation2.setPostTaskVersion(1); + + WorkflowTaskRelation relation3 = new WorkflowTaskRelation(); + relation3.setPreTaskCode(2L); + relation3.setPreTaskVersion(1); + relation3.setPostTaskCode(4L); + relation3.setPostTaskVersion(1); + + WorkflowTaskRelation relation4 = new WorkflowTaskRelation(); + relation4.setPreTaskCode(3L); + relation4.setPreTaskVersion(1); + relation4.setPostTaskCode(4L); + relation4.setPostTaskVersion(1); + + WorkflowTaskRelation relation5 = new WorkflowTaskRelation(); + relation5.setPreTaskCode(4L); + relation5.setPreTaskVersion(1); + relation5.setPostTaskCode(3L); + relation5.setPostTaskVersion(1); + + workflowTaskRelations.add(relation1); + workflowTaskRelations.add(relation2); + workflowTaskRelations.add(relation3); + workflowTaskRelations.add(relation4); + workflowTaskRelations.add(relation5); + + Assertions.assertThrows(IllegalArgumentException.class, + () -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + } +} From 7cbfb43261f5223da0c76576485ca4ab0e080fe6 Mon Sep 17 00:00:00 2001 From: shangeyao Date: Fri, 26 Jun 2026 10:45:34 +0800 Subject: [PATCH 2/3] [Improvement-17057][Master] Fix CodeQL warning for ignored Queue.offer return value Use Deque.addLast/pollFirst instead of Queue.offer/poll in DAG validation. Co-authored-by: Cursor --- .../server/master/engine/graph/WorkflowGraph.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java index 067912ac4149..9e14e1574654 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java @@ -24,11 +24,11 @@ import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Deque; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Queue; import java.util.Set; import java.util.stream.Collectors; @@ -85,16 +85,16 @@ private void checkIfDAG(List workflowTaskRelations, List queue = new ArrayDeque<>(); + Deque queue = new ArrayDeque<>(); for (Map.Entry entry : inDegreeCount.entrySet()) { if (entry.getValue() == 0) { - queue.offer(entry.getKey()); + queue.addLast(entry.getKey()); } } Set sortedTaskCodes = new HashSet<>(); while (!queue.isEmpty()) { - Long taskCode = queue.poll(); + Long taskCode = queue.pollFirst(); sortedTaskCodes.add(taskCode); List postCodes = postTaskCodeMap.get(taskCode); @@ -104,7 +104,7 @@ private void checkIfDAG(List workflowTaskRelations, List Date: Sat, 27 Jun 2026 15:44:01 +0800 Subject: [PATCH 3/3] [Improvement-17057][Master] Include cyclic task details in DAG validation error Co-authored-by: Cursor --- .../master/engine/graph/WorkflowGraph.java | 18 +++++++++++- .../graph/WorkflowGraphCheckIfDAGTest.java | 28 ++++++++++--------- 2 files changed, 32 insertions(+), 14 deletions(-) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java index 9e14e1574654..6cdfaf9ae9ea 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java @@ -110,10 +110,26 @@ private void checkIfDAG(List workflowTaskRelations, List taskDefinitionByCode = taskDefinitions.stream() + .collect(Collectors.toMap(TaskDefinition::getCode, taskDefinition -> taskDefinition)); + List cyclicTasks = inDegreeCount.entrySet() + .stream() + .filter(entry -> entry.getValue() > 0) + .map(entry -> formatTaskIdentifier(entry.getKey(), taskDefinitionByCode)) + .collect(Collectors.toList()); + throw new IllegalArgumentException( + "The workflow task relation is not a DAG, cyclic tasks: " + String.join(", ", cyclicTasks)); } } + private String formatTaskIdentifier(Long taskCode, Map taskDefinitionByCode) { + TaskDefinition taskDefinition = taskDefinitionByCode.get(taskCode); + if (taskDefinition != null && taskDefinition.getName() != null && !taskDefinition.getName().isEmpty()) { + return taskDefinition.getName() + "(" + taskCode + ")"; + } + return String.valueOf(taskCode); + } + @Override public List getStartNodes() { return predecessors.entrySet() diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java index b4005038afb0..58436a45815b 100644 --- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java @@ -90,8 +90,10 @@ public void checkIfDAGCycleThrowsException() { workflowTaskRelations.add(relation1); workflowTaskRelations.add(relation2); - Assertions.assertThrows(IllegalArgumentException.class, + IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class, () -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + Assertions.assertTrue(exception.getMessage().contains("1")); + Assertions.assertTrue(exception.getMessage().contains("2")); } @Test @@ -172,9 +174,13 @@ public void checkIfDAGComplexDAGNoException() { @Test public void checkIfDAGComplexDAGThrowsException() { TaskDefinition task1 = new TaskDefinition(1L, 1); + task1.setName("task1"); TaskDefinition task2 = new TaskDefinition(2L, 1); + task2.setName("task2"); TaskDefinition task3 = new TaskDefinition(3L, 1); + task3.setName("task3"); TaskDefinition task4 = new TaskDefinition(4L, 1); + task4.setName("task4"); taskDefinitions.add(task1); taskDefinitions.add(task2); taskDefinitions.add(task3); @@ -187,36 +193,32 @@ public void checkIfDAGComplexDAGThrowsException() { relation1.setPostTaskVersion(1); WorkflowTaskRelation relation2 = new WorkflowTaskRelation(); - relation2.setPreTaskCode(1L); + relation2.setPreTaskCode(2L); relation2.setPreTaskVersion(1); relation2.setPostTaskCode(3L); relation2.setPostTaskVersion(1); WorkflowTaskRelation relation3 = new WorkflowTaskRelation(); - relation3.setPreTaskCode(2L); + relation3.setPreTaskCode(3L); relation3.setPreTaskVersion(1); relation3.setPostTaskCode(4L); relation3.setPostTaskVersion(1); WorkflowTaskRelation relation4 = new WorkflowTaskRelation(); - relation4.setPreTaskCode(3L); + relation4.setPreTaskCode(4L); relation4.setPreTaskVersion(1); - relation4.setPostTaskCode(4L); + relation4.setPostTaskCode(2L); relation4.setPostTaskVersion(1); - WorkflowTaskRelation relation5 = new WorkflowTaskRelation(); - relation5.setPreTaskCode(4L); - relation5.setPreTaskVersion(1); - relation5.setPostTaskCode(3L); - relation5.setPostTaskVersion(1); - workflowTaskRelations.add(relation1); workflowTaskRelations.add(relation2); workflowTaskRelations.add(relation3); workflowTaskRelations.add(relation4); - workflowTaskRelations.add(relation5); - Assertions.assertThrows(IllegalArgumentException.class, + IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class, () -> new WorkflowGraph(workflowTaskRelations, taskDefinitions)); + Assertions.assertTrue(exception.getMessage().contains("2")); + Assertions.assertTrue(exception.getMessage().contains("3")); + Assertions.assertTrue(exception.getMessage().contains("4")); } }