YARN-8179: Preemption does not happen due to natural_termination_factor when DRF is used. Contributed by Kyungwan Nam.

(cherry picked from commit 0b4c44bdeef62945b592d5761666ad026b629c0b)

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

@@ -198,8 +198,11 @@ public class PreemptableResourceCalculator
            */
           Resource resToObtain = qT.toBePreempted;
           if (!isReservedPreemptionCandidatesSelector) {
-            resToObtain = Resources.multiply(qT.toBePreempted,
-                context.getNaturalTerminationFactor());
+            if (Resources.greaterThan(rc, clusterResource, resToObtain,
+                Resource.newInstance(0, 0))) {
+              resToObtain = Resources.multiplyAndNormalizeUp(rc, qT.toBePreempted,
+                  context.getNaturalTerminationFactor(), Resource.newInstance(1, 1));
+            }
           }
 
           // Only add resToObtain when it >= 0

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

@@ -18,15 +18,28 @@
 
 package org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity;
 
+import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
+import org.junit.Before;
 import org.junit.Test;
 
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
 import static org.mockito.Matchers.argThat;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class TestProportionalCapacityPreemptionPolicyInterQueueWithDRF
     extends ProportionalCapacityPreemptionPolicyMockFramework {
+
+  @Before
+  public void setup() {
+    super.setup();
+    rc = new DominantResourceCalculator();
+    when(cs.getResourceCalculator()).thenReturn(rc);
+    policy = new ProportionalCapacityPreemptionPolicy(rmContext, cs, mClock);
+  }
+
   @Test
   public void testInterQueuePreemptionWithMultipleResource()
       throws Exception {
@@ -65,4 +78,47 @@ public class TestProportionalCapacityPreemptionPolicyInterQueueWithDRF
         new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
             getAppAttemptId(2))));
   }
+
+  @Test
+  public void testInterQueuePreemptionWithNaturalTerminationFactor()
+      throws Exception {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *       root
+     *      /   \
+     *     a     b
+     * </pre>
+     *
+     * Guaranteed resource of a/b are 50:50 Total cluster resource = 100
+     * Scenario: All resources are allocated to Queue A.
+     * Even though Queue B needs few resources like 1 VCore, some resources
+     * must be preempted from the app which is running in Queue A.
+     */
+
+    conf.setFloat(
+        CapacitySchedulerConfiguration.PREEMPTION_NATURAL_TERMINATION_FACTOR,
+        (float) 0.2);
+
+    String labelsConfig = "=100:50,true;";
+    String nodesConfig = // n1 has no label
+        "n1= res=100:50";
+    String queuesConfig =
+        // guaranteed,max,used,pending
+        "root(=[100:50 100:50 50:50 0:0]);" + // root
+            "-a(=[50:25 100:50 50:50 0:0]);" + // a
+            "-b(=[50:25 50:25 0:0 2:1]);"; // b
+
+    String appsConfig =
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t(1,2:1,n1,,50,false);"; // app1 in a
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    verify(mDisp, times(1)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(1))));
+  }
 }