Avoid incrementing app-missed-scheduling-opportunities when the queue limit is reached.

Author: TaoYang526

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractLeafQueue.java

@@ -1380,8 +1380,6 @@ public CSAssignment assignContainers(Resource clusterResource,
             parent.getQueuePath(), getQueuePath(), ActivityState.REJECTED,
             () -> ActivityDiagnosticConstant.QUEUE_DO_NOT_HAVE_ENOUGH_HEADROOM
                 + " from " + application.getApplicationId());
-        // Add missed scheduling opportunities for the application
-        application.addAppMissedSchedulingOpportunities();
         return assignment;
       } else{
         // If we don't allocate anything, and it is not skipped by application,

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerAppBackoff.java

@@ -19,6 +19,14 @@
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
 
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList;
+import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceSizing;
+import org.apache.hadoop.yarn.api.records.SchedulingRequest;
+import org.apache.hadoop.yarn.api.resource.PlacementConstraint;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.resourcemanager.MockAM;
 import org.apache.hadoop.yarn.server.resourcemanager.MockNM;
 import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
@@ -27,15 +35,19 @@
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.ResourceUsageMultiNodeLookupPolicy;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
 
+import static org.apache.hadoop.yarn.api.resource.PlacementConstraints.PlacementTargets.allocationTag;
+import static org.apache.hadoop.yarn.api.resource.PlacementConstraints.targetIn;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.APP_BACKOFF_MISSED_THRESHOLD;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.BACKOFF_ENABLED;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.APP_BACKOFF_INTERVAL_MS;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.DEFAULT_APP_BACKOFF_MISSED_THRESHOLD;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.DEFAULT_APP_BACKOFF_INTERVAL_MS;
+import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.MULTI_NODE_PLACEMENT_ENABLED;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.PREFIX;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerQueueHelpers.A1;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerQueueHelpers.A2;
@@ -109,86 +121,100 @@ public void testSchedulingWithAppBackoffEnabled() throws Exception {
     long appBackoffIntervalMs = 100L;
     CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
     setupQueueConfiguration(conf);
+    enabledMultiNodesPlacement(conf);
     conf.setAppBackoffEnabled(A1, true);
     conf.setAppBackoffIntervalMs(A1, appBackoffIntervalMs);
     conf.setAppBackoffMissedThreshold(A1, 3L);
 
-    // Register a node with 10GB memory
+    // Register a node
     MockRM rm = new MockRM(conf);
     rm.start();
     MockNM nm1 = rm.registerNode("127.0.0.1:1234", 10 * GB);
 
-    // Submit an application in queue B1 that requests 5GB memory
-    MockRMAppSubmissionData data1 =
-        MockRMAppSubmissionData.Builder.createWithMemory(5 * GB, rm)
-            .withAppName("app1")
-            .withUser("user")
-            .withAcls(null)
-            .withQueue(B1.getLeafName())
-            .withUnmanagedAM(false)
-            .build();
-    RMApp app1 = MockRMAppSubmitter.submit(rm, data1);
-    MockRM.launchAndRegisterAM(app1, rm, nm1);
-
-    // Submit an application in queue A1 that requests 2GB memory,
-    // available memory will be 3GB for the whole cluster.
+    // Submit an application in queue A1
     MockRMAppSubmissionData data =
         MockRMAppSubmissionData.Builder.createWithMemory(2 * GB, rm)
-            .withAppName("app2")
+            .withAppName("app1")
             .withUser("user")
             .withAcls(null)
             .withQueue(A1.getLeafName())
             .withUnmanagedAM(false)
             .build();
-    RMApp app2 = MockRMAppSubmitter.submit(rm, data);
-    MockAM am2 = MockRM.launchAndRegisterAM(app2, rm, nm1);
-
-    // Request 5GB memory for a container that cannot be satisfied
-    am2.allocate("*", 5 * GB, 1, new ArrayList<>());
+    RMApp app = MockRMAppSubmitter.submit(rm, data);
+    MockAM am = MockRM.launchAndRegisterAM(app, rm, nm1);
+
+    // Submit a request that cannot be satisfied due to the
+    // placement-constraint condition
+    PlacementConstraint pc = targetIn("node",
+        allocationTag("hbase-master")).build();
+    SchedulingRequest schedulingRequest = SchedulingRequest.newInstance(
+        1, Priority.newInstance(1), ExecutionTypeRequest.newInstance(), null,
+        ResourceSizing.newInstance(1, Resource.newInstance(2 * GB, 1)), pc);
+    am.addSchedulingRequest(ImmutableList.of(schedulingRequest));
+    am.doHeartbeat();
 
     CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
     AbstractLeafQueue queueA1 =
         (AbstractLeafQueue) cs.getQueue(A1.getLeafName());
-    FiCaSchedulerApp schedulerApp2 =
-        cs.getApplicationAttempt(am2.getApplicationAttemptId());
+    FiCaSchedulerApp schedulerApp =
+        cs.getApplicationAttempt(am.getApplicationAttemptId());
 
     // Simulate missed scheduling opportunities
     for (int i = 0; i < 3; i++) {
       cs.handle(new NodeUpdateSchedulerEvent(
           rm.getRMContext().getRMNodes().get(nm1.getNodeId())));
     }
-    assertFalse(queueA1.isAppInBackoffState(app2.getApplicationId()));
-    assertEquals(3L, schedulerApp2.getAppMissedSchedulingOpportunities());
+    assertFalse(queueA1.isAppInBackoffState(app.getApplicationId()));
+    assertEquals(3L, schedulerApp.getAppMissedSchedulingOpportunities());
 
     // Make the app enter backoff state when it reaches the missed threshold
     cs.handle(new NodeUpdateSchedulerEvent(
         rm.getRMContext().getRMNodes().get(nm1.getNodeId())));
 
-    // Verify the app is in backoff state
-    assertTrue(queueA1.isAppInBackoffState(app2.getApplicationId()));
-    assertEquals(0L, schedulerApp2.getAppMissedSchedulingOpportunities());
+    // Verify app is in backoff state
+    assertTrue(queueA1.isAppInBackoffState(app.getApplicationId()));
+    assertEquals(0L, schedulerApp.getAppMissedSchedulingOpportunities());
 
     // Wait for the backoff interval to expire
     GenericTestUtils.waitFor(
-        () -> !queueA1.isAppInBackoffState(app2.getApplicationId()),
+        () -> !queueA1.isAppInBackoffState(app.getApplicationId()),
         appBackoffIntervalMs, appBackoffIntervalMs * 2);
 
     // Verify app is no longer in backoff state after the backoff interval
-    assertFalse(queueA1.isAppInBackoffState(app2.getApplicationId()));
+    assertFalse(queueA1.isAppInBackoffState(app.getApplicationId()));
 
     // Simulate another missed scheduling opportunity
-    cs.handle(new NodeUpdateSchedulerEvent(rm.getRMContext().getRMNodes().get(nm1.getNodeId())));
-    assertFalse(queueA1.isAppInBackoffState(app2.getApplicationId()));
-    assertEquals(1L, schedulerApp2.getAppMissedSchedulingOpportunities());
+    cs.handle(new NodeUpdateSchedulerEvent(
+        rm.getRMContext().getRMNodes().get(nm1.getNodeId())));
+    assertFalse(queueA1.isAppInBackoffState(app.getApplicationId()));
+    assertEquals(1L, schedulerApp.getAppMissedSchedulingOpportunities());
 
-    // Kill app1 to release resource on nm1
-    rm.killApp(app1.getApplicationId());
+    // Request another request which can be allocated at first
+    am.allocate("*", 2 * GB, 1, new ArrayList<>());
+    cs.handle(new NodeUpdateSchedulerEvent(
+        rm.getRMContext().getRMNodes().get(nm1.getNodeId())));
 
-    // Now pending request should be allocated and not in backoff state
-    cs.handle(new NodeUpdateSchedulerEvent(rm.getRMContext().getRMNodes().get(nm1.getNodeId())));
-    assertFalse(queueA1.isAppInBackoffState(schedulerApp2.getApplicationId()));
-    assertEquals(0L, schedulerApp2.getAppMissedSchedulingOpportunities());
+    // new request should be allocated and app is not in backoff state
+    assertFalse(queueA1.isAppInBackoffState(schedulerApp.getApplicationId()));
+    assertEquals(0L, schedulerApp.getAppMissedSchedulingOpportunities());
 
     rm.stop();
   }
+
+  private void enabledMultiNodesPlacement(CapacitySchedulerConfiguration conf) {
+    conf.set(YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_HANDLER,
+        YarnConfiguration.SCHEDULER_RM_PLACEMENT_CONSTRAINTS_HANDLER);
+    conf.setBoolean(MULTI_NODE_PLACEMENT_ENABLED, true);
+    conf.setBoolean(PREFIX + MULTI_NODE_PLACEMENT_ENABLED, true);
+    conf.set(CapacitySchedulerConfiguration.MULTI_NODE_SORTING_POLICIES,
+        "resource-based");
+    conf.set(CapacitySchedulerConfiguration.MULTI_NODE_SORTING_POLICY_NAME,
+        "resource-based");
+    String policyName =
+        CapacitySchedulerConfiguration.MULTI_NODE_SORTING_POLICY_NAME +
+            ".resource-based.class";
+    conf.set(policyName, ResourceUsageMultiNodeLookupPolicy.class.getName());
+    conf.setBoolean(CapacitySchedulerConfiguration.MULTI_NODE_PLACEMENT_ENABLED,
+        true);
+  }
 }