commit 77389ac7705f280ae5a7a3c15fc4e6bea6e797c0
author Richard Startin <richard@startree.ai> Thu Apr 28 17:50:14 2022 +0100
committer GitHub <noreply@github.com> Thu Apr 28 09:50:14 2022 -0700
tree 24cab450bd94399b9de1b2874131c4aa41740fca
parent 41ddeaaa8af4d077865e0d862e5b168ed52eb0cc

fix issue #2064: bug where RuntimeJobDag.generateJobList could loop until parallelism is reached when in JobQueue mode (#2065)

helix-core/src/main/java/org/apache/helix/task/RuntimeJobDag.java

diff --git a/helix-core/src/main/java/org/apache/helix/task/RuntimeJobDag.java b/helix-core/src/main/java/org/apache/helix/task/RuntimeJobDag.java
index 24a17da..ac8d44b 100644
--- a/helix-core/src/main/java/org/apache/helix/task/RuntimeJobDag.java
+++ b/helix-core/src/main/java/org/apache/helix/task/RuntimeJobDag.java
@@ -198,12 +198,14 @@
     resetJobListAndDependencyMaps();
     computeIndependentNodes();
     _readyJobList.addAll(_independentNodes);
-    if (_isJobQueue && _readyJobList.size() > 0) {
+    if (_isJobQueue && !_readyJobList.isEmpty()) {
       // For job queue, only get number of parallel jobs to run in the ready list.
       for (int i = 1; i < _numParallelJobs; i++) {
-        if (_parentsToChildren.containsKey(_readyJobList.peekLast())) {
-          _readyJobList.offer(_parentsToChildren.get(_readyJobList.peekLast()).iterator().next());
+        Set<String> children = _parentsToChildren.get(_readyJobList.peekLast());
+        if (children == null) {
+          break;
         }
+        _readyJobList.offer(children.iterator().next());
       }
     }
     _hasDagChanged = false;

helix-core/src/test/java/org/apache/helix/integration/task/TestRuntimeJobDag.java

diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestRuntimeJobDag.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestRuntimeJobDag.java
index 14c1de2..b06ae87 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/TestRuntimeJobDag.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestRuntimeJobDag.java
@@ -24,11 +24,25 @@
 import java.util.List;
 import java.util.Set;
 
+import org.apache.helix.task.JobDag;
 import org.apache.helix.task.RuntimeJobDag;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
 public class TestRuntimeJobDag {
+
+  @Test
+  public void testBuildJobQueueWithParallelismExceedingJobCount() {
+    JobDag jobDag = new JobDag();
+    jobDag.addNode("parent");
+    jobDag.addParentToChild("parent", "child");
+    jobDag.addParentToChild("child", "grandchild");
+    RuntimeJobDag runtimeJobDag = new RuntimeJobDag(jobDag, true, Integer.MAX_VALUE, 1);
+    Assert.assertEquals(runtimeJobDag.getNextJob(), "parent");
+    Assert.assertEquals(runtimeJobDag.getNextJob(), "child");
+    Assert.assertEquals(runtimeJobDag.getNextJob(), "grandchild");
+  }
+
   private Set<String> actualJobs;
   private Set<String> expectedJobs;