Address review comments

Author: afchung

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/scheduler/OpportunisticContainersQueuePolicy.java

@@ -33,7 +33,7 @@ public enum OpportunisticContainersQueuePolicy {
    */
   BY_QUEUE_LEN,
   /**
-   * Determines wheether or not to run a container based on the amount of
+   * Determines whether or not to run a container based on the amount of
    * resource capacity the node has.
    * Sums up the resources running + already queued at the node, compares
    * it with the total capacity of the node, and accepts the new container only

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerSchedulerTest.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitor;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.scheduler.ContainerScheduler;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.scheduler.TestContainerSchedulerQueuing;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerStartContext;
 import org.slf4j.LoggerFactory;
@@ -53,6 +54,11 @@
 
 import static org.mockito.Mockito.spy;
 
+/**
+ * Base test class that overrides the behavior of
+ * {@link ContainerStateTransitionListener} for testing
+ * the {@link ContainerScheduler}.
+ */
 public class BaseContainerSchedulerTest extends BaseContainerManagerTest {
   private static final long TWO_GB = 2048 * 1024 * 1024L;
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/scheduler/TestContainerSchedulerOppContainersByResources.java

@@ -46,10 +46,10 @@
 import java.util.concurrent.TimeoutException;
 
 /**
- * Tests the behavior of {@link ContainerScheduler} when the max queue length
- * is set to {@literal < 0} such that the NM only queues
- * containers if there's enough resources on the node to start
- * all queued containers.
+ * Tests the behavior of {@link ContainerScheduler} when its queueing policy
+ * is set to {@link OpportunisticContainersQueuePolicy#BY_RESOURCES}
+ * such that the NM only queues containers if there's enough resources
+ * on the node to start all queued containers.
  */
 public class TestContainerSchedulerOppContainersByResources
     extends BaseContainerSchedulerTest {
@@ -62,9 +62,17 @@ public void setup() throws IOException {
     conf.set(YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_QUEUE_POLICY,
         OpportunisticContainersQueuePolicy.BY_RESOURCES.name());
     super.setup();
+    containerManager.start();
   }
 
-  private static boolean isSuccessfulRun(final ContainerStatus containerStatus) {
+  /**
+   * Checks if a container is in a running or successfully run state.
+   * @param containerStatus the container status
+   * @return true if the container is running or completed
+   * with a successful state, false if the container has not started or failed
+   */
+  private static boolean isContainerInSuccessfulState(
+      final ContainerStatus containerStatus) {
     final org.apache.hadoop.yarn.api.records.ContainerState state =
         containerStatus.getState();
     final ContainerSubState subState = containerStatus.getContainerSubState();
@@ -73,6 +81,8 @@ private static boolean isSuccessfulRun(final ContainerStatus containerStatus) {
     case COMPLETING:
     case DONE:
       if (subState == ContainerSubState.DONE) {
+        // If the state is not COMPLETE, then the
+        // container is a failed container
         return state ==
             org.apache.hadoop.yarn.api.records.ContainerState.COMPLETE;
       }
@@ -106,7 +116,7 @@ private void verifyRunAndKilledContainers(
 
           for (final ContainerStatus status : containerStatuses) {
             if (runContainers.contains(status.getContainerId())) {
-              if (!isSuccessfulRun(status)) {
+              if (!isContainerInSuccessfulState(status)) {
                 return false;
               }
             } else if (killedContainers.contains(status.getContainerId())) {
@@ -124,62 +134,10 @@ private void verifyRunAndKilledContainers(
   }
 
   /**
-   * Tests that newly arrived containers after the resources are filled up
-   * get killed and never get killed.
+   * Verifies that nothing is queued at the container scheduler.
    */
-  @Test
-  public void testOpportunisticRunsWhenResourcesAvailable() throws Exception {
-    containerManager.start();
-    List<StartContainerRequest> list = new ArrayList<>();
-    final int numContainers = 8;
-    final int numContainersQueued = 4;
-    final Set<ContainerId> runContainers = new HashSet<>();
-    final Set<ContainerId> killedContainers = new HashSet<>();
-
-    for (int i = 0; i < numContainers; i++) {
-      // OContainers that should be run
-      list.add(StartContainerRequest.newInstance(
-          recordFactory.newRecordInstance(ContainerLaunchContext.class),
-          createContainerToken(createContainerId(i), DUMMY_RM_IDENTIFIER,
-              context.getNodeId(),
-              user, BuilderUtils.newResource(512, 1),
-              context.getContainerTokenSecretManager(), null,
-              ExecutionType.OPPORTUNISTIC)));
-    }
-
-    StartContainersRequest allRequests =
-        StartContainersRequest.newInstance(list);
-    containerManager.startContainers(allRequests);
-
-    // Wait for containers to start
-    for (int i = 0; i < numContainersQueued; i++) {
-      final ContainerId containerId = createContainerId(i);
-      BaseContainerManagerTest
-          .waitForNMContainerState(containerManager, containerId,
-              ContainerState.RUNNING, 40);
-      runContainers.add(containerId);
-    }
-
-    // Wait for containers to be killed
-    for (int i = numContainersQueued; i < numContainers; i++) {
-      final ContainerId containerId = createContainerId(i);
-      BaseContainerManagerTest
-          .waitForNMContainerState(containerManager, createContainerId(i),
-              ContainerState.DONE, 40);
-      killedContainers.add(containerId);
-    }
-
-    Thread.sleep(5000);
-
-    // Get container statuses.
-    List<ContainerId> statList = new ArrayList<>();
-    for (int i = 0; i < numContainers; i++) {
-      statList.add(createContainerId(i));
-    }
-
-    verifyRunAndKilledContainers(
-        statList, numContainers, runContainers, killedContainers);
-
+  private void verifyNothingQueued() {
+    // Check that nothing is queued
     ContainerScheduler containerScheduler =
         containerManager.getContainerScheduler();
     Assert.assertEquals(0,
@@ -194,19 +152,15 @@ public void testOpportunisticRunsWhenResourcesAvailable() throws Exception {
   }
 
   /**
-   * Sets the max queue length to negative such that the NM only queues
-   * containers if there's enough resources on the node to start
-   * all queued containers.
    * Tests that newly arrived containers after the resources are filled up
-   * get killed and never get killed.
+   * get killed and never gets run.
    */
   @Test
   public void testKillOpportunisticWhenNoResourcesAvailable() throws Exception {
-    containerManager.start();
-    List<StartContainerRequest> list = new ArrayList<>();
+    List<StartContainerRequest> startContainerRequests = new ArrayList<>();
 
     // GContainer that takes up the whole node
-    list.add(StartContainerRequest.newInstance(
+    startContainerRequests.add(StartContainerRequest.newInstance(
         recordFactory.newRecordInstance(ContainerLaunchContext.class),
         createContainerToken(createContainerId(0), DUMMY_RM_IDENTIFIER,
             context.getNodeId(),
@@ -215,7 +169,7 @@ public void testKillOpportunisticWhenNoResourcesAvailable() throws Exception {
             ExecutionType.GUARANTEED)));
 
     // OContainer that should be killed
-    list.add(StartContainerRequest.newInstance(
+    startContainerRequests.add(StartContainerRequest.newInstance(
         recordFactory.newRecordInstance(ContainerLaunchContext.class),
         createContainerToken(createContainerId(1), DUMMY_RM_IDENTIFIER,
             context.getNodeId(),
@@ -224,7 +178,7 @@ public void testKillOpportunisticWhenNoResourcesAvailable() throws Exception {
             ExecutionType.OPPORTUNISTIC)));
 
     StartContainersRequest allRequests =
-        StartContainersRequest.newInstance(list);
+        StartContainersRequest.newInstance(startContainerRequests);
     containerManager.startContainers(allRequests);
 
     BaseContainerManagerTest.waitForNMContainerState(containerManager,
@@ -245,16 +199,67 @@ public void testKillOpportunisticWhenNoResourcesAvailable() throws Exception {
         Collections.singleton(createContainerId(1))
     );
 
-    ContainerScheduler containerScheduler =
-        containerManager.getContainerScheduler();
-    Assert.assertEquals(0,
-        containerScheduler.getNumQueuedContainers());
-    Assert.assertEquals(0,
-        containerScheduler.getNumQueuedGuaranteedContainers());
-    Assert.assertEquals(0,
-        containerScheduler.getNumQueuedOpportunisticContainers());
-    Assert.assertEquals(0,
-        metrics.getQueuedOpportunisticContainers());
-    Assert.assertEquals(0, metrics.getQueuedGuaranteedContainers());
+    verifyNothingQueued();
+  }
+
+  /**
+   * Tests that newly arrived containers after the resources are filled up
+   * get killed and never gets run.
+   * This scenario is more granular and runs more small container compared to
+   * {@link #testKillOpportunisticWhenNoResourcesAvailable()}.
+   */
+  @Test
+  public void testOpportunisticRunsWhenResourcesAvailable() throws Exception {
+    List<StartContainerRequest> startContainerRequests = new ArrayList<>();
+    final int numContainers = 8;
+    final int numContainersQueued = 4;
+    final Set<ContainerId> runContainers = new HashSet<>();
+    final Set<ContainerId> killedContainers = new HashSet<>();
+
+    for (int i = 0; i < numContainers; i++) {
+      // OContainers that should be run
+      startContainerRequests.add(StartContainerRequest.newInstance(
+          recordFactory.newRecordInstance(ContainerLaunchContext.class),
+          createContainerToken(createContainerId(i), DUMMY_RM_IDENTIFIER,
+              context.getNodeId(),
+              user, BuilderUtils.newResource(512, 1),
+              context.getContainerTokenSecretManager(), null,
+              ExecutionType.OPPORTUNISTIC)));
+    }
+
+    StartContainersRequest allRequests =
+        StartContainersRequest.newInstance(startContainerRequests);
+    containerManager.startContainers(allRequests);
+
+    // Wait for containers to start
+    for (int i = 0; i < numContainersQueued; i++) {
+      final ContainerId containerId = createContainerId(i);
+      BaseContainerManagerTest
+          .waitForNMContainerState(containerManager, containerId,
+              ContainerState.RUNNING, 40);
+      runContainers.add(containerId);
+    }
+
+    // Wait for containers to be killed
+    for (int i = numContainersQueued; i < numContainers; i++) {
+      final ContainerId containerId = createContainerId(i);
+      BaseContainerManagerTest
+          .waitForNMContainerState(containerManager, createContainerId(i),
+              ContainerState.DONE, 40);
+      killedContainers.add(containerId);
+    }
+
+    Thread.sleep(5000);
+
+    // Get container statuses.
+    List<ContainerId> statList = new ArrayList<>();
+    for (int i = 0; i < numContainers; i++) {
+      statList.add(createContainerId(i));
+    }
+
+    verifyRunAndKilledContainers(
+        statList, numContainers, runContainers, killedContainers);
+
+    verifyNothingQueued();
   }
 }