[helix] branch master updated: fix issue #2064: bug where RuntimeJobDag.generateJobList could loop until parallelism is reached when in JobQueue mode (#2065)

[jxue]

This is an automated email from the ASF dual-hosted git repository.

jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git


The following commit(s) were added to refs/heads/master by this push:
     new 77389ac77 fix issue #2064: bug where RuntimeJobDag.generateJobList 
could loop until parallelism is reached when in JobQueue mode (#2065)
77389ac77 is described below

commit 77389ac7705f280ae5a7a3c15fc4e6bea6e797c0
Author: Richard Startin <rich...@startree.ai>
AuthorDate: Thu Apr 28 17:50:14 2022 +0100

    fix issue #2064: bug where RuntimeJobDag.generateJobList could loop until 
parallelism is reached when in JobQueue mode (#2065)
---
 .../src/main/java/org/apache/helix/task/RuntimeJobDag.java |  8 +++++---
 .../apache/helix/integration/task/TestRuntimeJobDag.java   | 14 ++++++++++++++
 2 files changed, 19 insertions(+), 3 deletions(-)

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 24a17da8f..ac8d44bd4 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 @@ public class RuntimeJobDag extends JobDag {
     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;
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 14c1de2b6..b06ae8785 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.HashSet;
 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;