YARN-4148. When killing app, RM releases app's resource before they are released by NM. Contributed by Jason Lowe.

(cherry picked from commit dbac88baa9a15eaa8fcd0b464172d5cd9006199c)

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

@@ -257,6 +257,7 @@ public abstract class AbstractYarnScheduler
     }
 
     application.containerLaunchedOnNode(containerId, node.getNodeID());
+    node.containerStarted(containerId);
   }
   
   protected void containerIncreasedOnNode(ContainerId containerId,

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

@@ -65,8 +65,8 @@ public abstract class SchedulerNode {
       ResourceUtilization.newInstance(0, 0, 0f);
 
   /* set of containers that are allocated containers */
-  protected final Map<ContainerId, RMContainer> launchedContainers =
-      new HashMap<ContainerId, RMContainer>();
+  private final Map<ContainerId, ContainerInfo> launchedContainers =
+      new HashMap<>();
 
   private final RMNode rmNode;
   private final String nodeName;
@@ -148,12 +148,24 @@ public abstract class SchedulerNode {
    * @param rmContainer
    *          allocated container
    */
-  public synchronized void allocateContainer(RMContainer rmContainer) {
+  public void allocateContainer(RMContainer rmContainer) {
+    allocateContainer(rmContainer, false);
+  }
+
+  /**
+   * The Scheduler has allocated containers on this node to the given
+   * application.
+   * @param rmContainer Allocated container
+   * @param launchedOnNode True if the container has been launched
+   */
+  private synchronized void allocateContainer(RMContainer rmContainer,
+      boolean launchedOnNode) {
     Container container = rmContainer.getContainer();
     deductAvailableResource(container.getResource());
     ++numContainers;
 
-    launchedContainers.put(container.getId(), rmContainer);
+    launchedContainers.put(container.getId(),
+        new ContainerInfo(rmContainer, launchedOnNode));
 
     LOG.info("Assigned container " + container.getId() + " of capacity "
         + container.getResource() + " on host " + rmNode.getNodeAddress()
@@ -236,20 +248,25 @@ public abstract class SchedulerNode {
   /**
    * Release an allocated container on this node.
    * 
-   * @param container
-   *          container to be released
+   * @param containerId ID of container to be released.
+   * @param releasedByNode whether the release originates from a node update.
    */
-  public synchronized void releaseContainer(Container container) {
-    if (!isValidContainer(container.getId())) {
-      LOG.error("Invalid container released " + container);
+  public synchronized void releaseContainer(ContainerId containerId,
+      boolean releasedByNode) {
+    ContainerInfo info = launchedContainers.get(containerId);
+    if (info == null) {
       return;
     }
 
-    /* remove the containers from the nodemanger */
-    if (null != launchedContainers.remove(container.getId())) {
-      updateResourceForReleasedContainer(container);
+    if (!releasedByNode && info.launchedOnNode) {
+      // wait until node reports container has completed
+      return;
     }
 
+    launchedContainers.remove(containerId);
+    Container container = info.container.getContainer();
+    updateResourceForReleasedContainer(container);
+
     LOG.info("Released container " + container.getId() + " of capacity "
         + container.getResource() + " on host " + rmNode.getNodeAddress()
         + ", which currently has " + numContainers + " containers, "
@@ -257,6 +274,17 @@ public abstract class SchedulerNode {
         + " available" + ", release resources=" + true);
   }
 
+   /**
+   * Inform the node that a container has launched.
+   * @param containerId ID of the launched container
+   */
+  public synchronized void containerStarted(ContainerId containerId) {
+    ContainerInfo info = launchedContainers.get(containerId);
+    if (info != null) {
+      info.launchedOnNode = true;
+    }
+  }
+
   private synchronized void addAvailableResource(Resource resource) {
     if (resource == null) {
       LOG.error("Invalid resource addition of null resource for "
@@ -305,7 +333,25 @@ public abstract class SchedulerNode {
   }
 
   public synchronized List<RMContainer> getCopiedListOfRunningContainers() {
-    return new ArrayList<RMContainer>(launchedContainers.values());
+    List<RMContainer> result = new ArrayList<>(launchedContainers.size());
+    for (ContainerInfo info : launchedContainers.values()) {
+      result.add(info.container);
+    }
+    return result;
+  }
+
+  /**
+   * Get the container for the specified container ID.
+   * @param containerId The container ID
+   * @return The container for the specified container ID
+   */
+  protected synchronized RMContainer getContainer(ContainerId containerId) {
+    RMContainer container = null;
+    ContainerInfo info = launchedContainers.get(containerId);
+    if (info != null) {
+      container = info.container;
+    }
+    return container;
   }
 
   public synchronized RMContainer getReservedContainer() {
@@ -321,7 +367,7 @@ public abstract class SchedulerNode {
     if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
       return;
     }
-    allocateContainer(rmContainer);
+    allocateContainer(rmContainer, true);
   }
   
   public Set<String> getLabels() {
@@ -377,4 +423,15 @@ public abstract class SchedulerNode {
   public ResourceUtilization getNodeUtilization() {
     return this.nodeUtilization;
   }
+
+
+  private static class ContainerInfo {
+    private final RMContainer container;
+    private boolean launchedOnNode;
+
+    public ContainerInfo(RMContainer container, boolean launchedOnNode) {
+      this.container = container;
+      this.launchedOnNode = launchedOnNode;
+    }
+  }
 }

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

@@ -1074,6 +1074,7 @@ public class CapacityScheduler extends
       RMContainer container = getRMContainer(containerId);
       super.completedContainer(container, completedContainer,
         RMContainerEventType.FINISHED);
+      node.releaseContainer(containerId, true);
       if (container != null) {
         releasedContainers++;
         Resource rs = container.getAllocatedResource();

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

@@ -1451,7 +1451,7 @@ public class LeafQueue extends AbstractCSQueue {
               application.containerCompleted(rmContainer, containerStatus,
                   event, node.getPartition());
 
-          node.releaseContainer(container);
+          node.releaseContainer(rmContainer.getContainerId(), false);
         }
 
         // Book-keeping

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerNode.java

@@ -125,7 +125,7 @@ public class FiCaSchedulerNode extends SchedulerNode {
 
   // According to decisions from preemption policy, mark the container to killable
   public synchronized void markContainerToKillable(ContainerId containerId) {
-    RMContainer c = launchedContainers.get(containerId);
+    RMContainer c = getContainer(containerId);
     if (c != null && !killableContainers.containsKey(containerId)) {
       killableContainers.put(containerId, c);
       Resources.addTo(totalKillableResources, c.getAllocatedResource());
@@ -135,7 +135,7 @@ public class FiCaSchedulerNode extends SchedulerNode {
   // According to decisions from preemption policy, mark the container to
   // non-killable
   public synchronized void markContainerToNonKillable(ContainerId containerId) {
-    RMContainer c = launchedContainers.get(containerId);
+    RMContainer c = getContainer(containerId);
     if (c != null && killableContainers.containsKey(containerId)) {
       killableContainers.remove(containerId);
       Resources.subtractFrom(totalKillableResources, c.getAllocatedResource());

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java

@@ -871,7 +871,7 @@ public class FairScheduler extends
       application.unreserve(rmContainer.getReservedPriority(), node);
     } else {
       application.containerCompleted(rmContainer, containerStatus, event);
-      node.releaseContainer(container);
+      node.releaseContainer(rmContainer.getContainerId(), false);
       updateRootQueueMetrics();
     }
 
@@ -1053,6 +1053,7 @@ public class FairScheduler extends
       LOG.debug("Container FINISHED: " + containerId);
       super.completedContainer(getRMContainer(containerId),
           completedContainer, RMContainerEventType.FINISHED);
+      node.releaseContainer(containerId, true);
     }
 
     // If the node is decommissioning, send an update to have the total

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java

@@ -746,6 +746,7 @@ public class FifoScheduler extends
       LOG.debug("Container FINISHED: " + containerId);
       super.completedContainer(getRMContainer(containerId),
           completedContainer, RMContainerEventType.FINISHED);
+      node.releaseContainer(containerId, true);
     }
 
     // Updating node resource utilization
@@ -917,7 +918,7 @@ public class FifoScheduler extends
         RMNodeLabelsManager.NO_LABEL);
 
     // Inform the node
-    node.releaseContainer(container);
+    node.releaseContainer(rmContainer.getContainerId(), false);
     
     // Update total usage
     Resources.subtractFrom(usedResource, container.getResource());

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

@@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerState;
+import org.apache.hadoop.yarn.api.records.ContainerStatus;
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceOption;
@@ -436,6 +437,119 @@ public class TestAbstractYarnScheduler extends ParameterizedSchedulerTestBase {
     }
   }
 
+  @Test(timeout=60000)
+  public void testContainerReleasedByNode() throws Exception {
+    System.out.println("Starting testContainerReleasedByNode");
+    configureScheduler();
+    YarnConfiguration conf = getConf();
+    MockRM rm1 = new MockRM(conf);
+    try {
+      rm1.start();
+      RMApp app1 =
+          rm1.submitApp(200, "name", "user",
+              new HashMap<ApplicationAccessType, String>(), false, "default",
+              -1, null, "Test", false, true);
+      MockNM nm1 =
+          new MockNM("127.0.0.1:1234", 10240, rm1.getResourceTrackerService());
+      nm1.registerNode();
+
+      MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
+
+      // allocate a container that fills more than half the node
+      am1.allocate("127.0.0.1", 8192, 1, new ArrayList<ContainerId>());
+      nm1.nodeHeartbeat(true);
+
+      // wait for containers to be allocated.
+      List<Container> containers =
+          am1.allocate(new ArrayList<ResourceRequest>(),
+              new ArrayList<ContainerId>()).getAllocatedContainers();
+      while (containers.isEmpty()) {
+        Thread.sleep(10);
+        nm1.nodeHeartbeat(true);
+        containers = am1.allocate(new ArrayList<ResourceRequest>(),
+            new ArrayList<ContainerId>()).getAllocatedContainers();
+      }
+
+      // release the container from the AM
+      ContainerId cid = containers.get(0).getId();
+      List<ContainerId> releasedContainers = new ArrayList<>(1);
+      releasedContainers.add(cid);
+      List<ContainerStatus> completedContainers = am1.allocate(
+          new ArrayList<ResourceRequest>(), releasedContainers)
+          .getCompletedContainersStatuses();
+      while (completedContainers.isEmpty()) {
+        Thread.sleep(10);
+        completedContainers = am1.allocate(
+          new ArrayList<ResourceRequest>(), releasedContainers)
+          .getCompletedContainersStatuses();
+      }
+
+      // verify new container can be allocated immediately because container
+      // never launched on the node
+      containers = am1.allocate("127.0.0.1", 8192, 1,
+          new ArrayList<ContainerId>()).getAllocatedContainers();
+      nm1.nodeHeartbeat(true);
+      while (containers.isEmpty()) {
+        Thread.sleep(10);
+        nm1.nodeHeartbeat(true);
+        containers = am1.allocate(new ArrayList<ResourceRequest>(),
+            new ArrayList<ContainerId>()).getAllocatedContainers();
+      }
+
+      // launch the container on the node
+      cid = containers.get(0).getId();
+      nm1.nodeHeartbeat(cid.getApplicationAttemptId(), cid.getContainerId(),
+          ContainerState.RUNNING);
+      rm1.waitForState(nm1, cid, RMContainerState.RUNNING);
+
+      // release the container from the AM
+      releasedContainers.clear();
+      releasedContainers.add(cid);
+      completedContainers = am1.allocate(
+          new ArrayList<ResourceRequest>(), releasedContainers)
+          .getCompletedContainersStatuses();
+      while (completedContainers.isEmpty()) {
+        Thread.sleep(10);
+        completedContainers = am1.allocate(
+          new ArrayList<ResourceRequest>(), releasedContainers)
+          .getCompletedContainersStatuses();
+      }
+
+      // verify new container cannot be allocated immediately because container
+      // has not been released by the node
+      containers = am1.allocate("127.0.0.1", 8192, 1,
+          new ArrayList<ContainerId>()).getAllocatedContainers();
+      nm1.nodeHeartbeat(true);
+      Assert.assertTrue("new container allocated before node freed old",
+          containers.isEmpty());
+      for (int i = 0; i < 10; ++i) {
+        Thread.sleep(10);
+        containers = am1.allocate(new ArrayList<ResourceRequest>(),
+            new ArrayList<ContainerId>()).getAllocatedContainers();
+        nm1.nodeHeartbeat(true);
+        Assert.assertTrue("new container allocated before node freed old",
+            containers.isEmpty());
+      }
+
+      // free the old container from the node
+      nm1.nodeHeartbeat(cid.getApplicationAttemptId(), cid.getContainerId(),
+          ContainerState.COMPLETE);
+
+      // verify new container is now allocated
+      containers = am1.allocate(new ArrayList<ResourceRequest>(),
+          new ArrayList<ContainerId>()).getAllocatedContainers();
+      while (containers.isEmpty()) {
+        Thread.sleep(10);
+        nm1.nodeHeartbeat(true);
+        containers = am1.allocate(new ArrayList<ResourceRequest>(),
+            new ArrayList<ContainerId>()).getAllocatedContainers();
+      }
+    } finally {
+      rm1.stop();
+      System.out.println("Stopping testContainerReleasedByNode");
+    }
+  }
+
   @Test(timeout = 60000)
   public void testResourceRequestRestoreWhenRMContainerIsAtAllocated()
       throws Exception {

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
 
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doNothing;
@@ -159,7 +160,8 @@ public class TestChildQueueOrder {
     }).
     when(queue).assignContainers(eq(clusterResource), eq(node), 
         any(ResourceLimits.class), any(SchedulingMode.class));
-    doNothing().when(node).releaseContainer(any(Container.class));
+    doNothing().when(node).releaseContainer(any(ContainerId.class),
+        anyBoolean());
   }
 
 
@@ -230,7 +232,8 @@ public class TestChildQueueOrder {
 
     FiCaSchedulerNode node_0 = 
       TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB);
-    doNothing().when(node_0).releaseContainer(any(Container.class));
+    doNothing().when(node_0).releaseContainer(any(ContainerId.class),
+        anyBoolean());
     
     final Resource clusterResource = 
       Resources.createResource(numNodes * (memoryPerNode*GB),