YARN-6538. Inter Queue preemption is not happening when DRF is configured in extreme resource scale cases.

Author: brumi1024

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java

@@ -153,6 +153,10 @@ public boolean isAnyMajorResourceZeroOrNegative(Resource resource) {
     return resource.getMemorySize() <= 0;
   }
 
+  @Override
+  public boolean isAnyRequestedResourceZeroOrNegative(Resource available, Resource resource) {
+    return resource.getMemorySize() == 0;
+  }
   @Override
   public boolean isAnyMajorResourceAboveZero(Resource resource) {
     return resource.getMemorySize() > 0;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java

@@ -642,6 +642,22 @@ public boolean isAnyMajorResourceZeroOrNegative(Resource resource) {
     return false;
   }
 
+  @Override
+  public boolean isAnyRequestedResourceZeroOrNegative(Resource available, Resource resource) {
+    int maxLength = ResourceUtils.getNumberOfCountableResourceTypes();
+    for (int i = 0; i < maxLength; i++) {
+      ResourceInformation resourceInformation = available.getResourceInformation(
+          i);
+      if (resourceInformation.getValue() != 0L) {
+        ResourceInformation ri2 = resource.getResourceInformation(i);
+        if (ri2.getValue() <= 0L) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+
   @Override
   public boolean isAnyMajorResourceAboveZero(Resource resource) {
     int maxLength = ResourceUtils.getNumberOfCountableResourceTypes();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java

@@ -278,6 +278,16 @@ public abstract float divide(
    */
   public abstract boolean isAnyMajorResourceZeroOrNegative(Resource resource);
 
+  /**
+   * Check if resource has any available (those that have non-zero available value)
+   * major resource types (which are all NodeManagers included) a zero value or negative value.
+   *
+   * @param available available resource
+   * @param resource resource
+   * @return returns true if any resource is zero.
+   */
+  public abstract boolean isAnyRequestedResourceZeroOrNegative(Resource available, Resource resource);
+
   /**
    * Get resource <code>r</code>and normalize down using step-factor
    * <code>stepFactor</code>.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/AbstractPreemptableResourceCalculator.java

@@ -224,7 +224,11 @@ protected void computeFixpointAllocation(Resource totGuarant,
 
       for (Iterator<TempQueuePerPartition> i = underserved.iterator(); i
           .hasNext();) {
-        if (!rc.isAnyMajorResourceAboveZero(unassigned)) {
+        // Exit the loop once any of the unassigned resources reach zero, except the optional
+        // ones which the queue has no guarantee on. This should ensure that extreme cases
+        // where one of the resources are significantly larger than the other (i.e: way below 1
+        // vcore per GB of memory)
+        if (rc.isAnyRequestedResourceZeroOrNegative(totGuarant, unassigned)) {
           break;
         }
 
@@ -266,7 +270,6 @@ protected void computeFixpointAllocation(Resource totGuarant,
     }
   }
 
-
   /**
    * This method is visible to allow sub-classes to override the initialization
    * behavior.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java

@@ -31,6 +31,7 @@
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue;
 import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.ResourceUtils;
+import org.mockito.Matchers;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
@@ -89,6 +90,56 @@ public void testInterQueuePreemptionWithMultipleResource() throws Exception {
             getAppAttemptId(2))));
   }
 
+  @Test
+  public void testInterQueuePreemptionWithMultipleResourceUnbalancedCase() throws IOException {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *           root
+     *           /  \
+     *          a    b
+     * </pre>
+     *
+     * Scenario: Queue B : Queue A has 90/10% capacity share.
+     * Almost all the resources are allocated to Queue A, but based on memory alone
+     * containers still can be allocated elsewhere. The app on A requests a lot more.
+     * The app on Queue B needs a little memory (still fits in the remaining  resources),
+     * but needs a lot of vcores. Enough resources must be preempted from the app on A
+     * to allow the app on B launch.
+     */
+    String labelsConfig = "=65536:50,true;"; // default partition
+    String nodesConfig = // only one node
+        "n1=";
+
+    String queuesConfig =
+        //    guaranteed, max,  used,   pending
+        "root(=[65536:50 65536:50 61440:45 132096:525]);" + // root
+            "-a(=[2048:5 65536:50 61440:45 131072:500]);" + // a
+            "-b(=[63488:45 65536:50 0:0 1024:25]);"; // b
+
+
+    String appsConfig =
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t(1,61440:45,n1,,20,false);" + // app1 in a
+            "b\t(1,1024:25,n1,,30,false)"; // app2 in b
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig, true);
+    Resource ul = Resource.newInstance(65536, 50);
+    when(((LeafQueue)(cs.getQueue("root.a")))
+        .getResourceLimitForAllUsers(any(), any(), any(), any())
+    ).thenReturn(ul);
+    policy.editSchedule();
+
+    // Preemption should happen in Queue a, preempt to Queue b
+    verify(eventHandler, times(1)).handle(Matchers.argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(1))));
+    verify(eventHandler, never()).handle(Matchers.argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(2))));
+  }
+
   @Test
   public void testInterQueuePreemptionWithNaturalTerminationFactor()
       throws Exception {