[Backport 2.x] Validate that all node IDs are unique (#192)

Validate that all node IDs are unique (#191) * Validate that all node IDs are unique * Use node map instead of a throwaway set * Remove unneeded diff argument since ignoreExitValue is used --------- (cherry picked from commit 0a30431) Signed-off-by: Daniel Widdis <[email protected]> Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
opensearch-project · Nov 22, 2023 · db30b87 · db30b87
1 parent cc2e348
commit db30b87
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 7 deletions.
diff --git a/build.gradle b/build.gradle
@@ -146,7 +146,8 @@ dependencies {
     configurations.all {
         resolutionStrategy {
             force("com.google.guava:guava:32.1.3-jre") // CVE for 31.1
-            force("org.eclipse.platform:org.eclipse.core.runtime:3.29.0") // CVE for 3.26.100
+            force("org.eclipse.platform:org.eclipse.core.runtime:3.29.0") // CVE for < 3.29.0
+            force("com.fasterxml.jackson.core:jackson-core:2.16.0") // Dependency Jar Hell
         }
     }
 }
@@ -239,7 +240,7 @@ diffCoverageReport {
             file.withOutputStream { out ->
                 exec {
                     ignoreExitValue true
-                    commandLine 'git', 'diff', '--no-color', '--minimal', diffBase, '|| true'
+                    commandLine 'git', 'diff', '--no-color', '--minimal', diffBase
                     standardOutput = out
                 }
             }

diff --git a/src/main/java/org/opensearch/flowframework/workflow/WorkflowProcessSorter.java b/src/main/java/org/opensearch/flowframework/workflow/WorkflowProcessSorter.java
@@ -29,7 +29,6 @@
 import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -158,14 +157,20 @@ private TimeValue parseTimeout(WorkflowNode node) {
 
     private static List<WorkflowNode> topologicalSort(List<WorkflowNode> workflowNodes, List<WorkflowEdge> workflowEdges) {
         // Basic validation
-        Set<String> nodeIds = workflowNodes.stream().map(n -> n.id()).collect(Collectors.toSet());
+        Map<String, WorkflowNode> nodeMap = new HashMap<>();
+        for (WorkflowNode node : workflowNodes) {
+            if (nodeMap.containsKey(node.id())) {
+                throw new FlowFrameworkException("Duplicate node id " + node.id() + ".", RestStatus.BAD_REQUEST);
+            }
+            nodeMap.put(node.id(), node);
+        }
         for (WorkflowEdge edge : workflowEdges) {
             String source = edge.source();
-            if (!nodeIds.contains(source)) {
+            if (!nodeMap.containsKey(source)) {
                 throw new FlowFrameworkException("Edge source " + source + " does not correspond to a node.", RestStatus.BAD_REQUEST);
             }
             String dest = edge.destination();
-            if (!nodeIds.contains(dest)) {
+            if (!nodeMap.containsKey(dest)) {
                 throw new FlowFrameworkException("Edge destination " + dest + " does not correspond to a node.", RestStatus.BAD_REQUEST);
             }
             if (source.equals(dest)) {
@@ -176,7 +181,6 @@ private static List<WorkflowNode> topologicalSort(List<WorkflowNode> workflowNod
         // Build predecessor and successor maps
         Map<WorkflowNode, Set<WorkflowEdge>> predecessorEdges = new HashMap<>();
         Map<WorkflowNode, Set<WorkflowEdge>> successorEdges = new HashMap<>();
-        Map<String, WorkflowNode> nodeMap = workflowNodes.stream().collect(Collectors.toMap(WorkflowNode::id, Function.identity()));
         for (WorkflowEdge edge : workflowEdges) {
             WorkflowNode source = nodeMap.get(edge.source());
             WorkflowNode dest = nodeMap.get(edge.destination());

diff --git a/src/test/java/org/opensearch/flowframework/workflow/WorkflowProcessSorterTests.java b/src/test/java/org/opensearch/flowframework/workflow/WorkflowProcessSorterTests.java
@@ -219,6 +219,10 @@ public void testExceptions() throws IOException {
         );
         assertEquals("Workflow step type [unimplemented_step] is not implemented.", ex.getMessage());
         assertEquals(RestStatus.NOT_IMPLEMENTED, ((FlowFrameworkException) ex).getRestStatus());
+
+        ex = assertThrows(FlowFrameworkException.class, () -> parse(workflow(List.of(node("A"), node("A")), Collections.emptyList())));
+        assertEquals("Duplicate node id A.", ex.getMessage());
+        assertEquals(RestStatus.BAD_REQUEST, ((FlowFrameworkException) ex).getRestStatus());
     }
 
     public void testSuccessfulGraphValidation() throws Exception {