MAPREDUCE-3780. Fixed a bug where applications killed before getting activated were not getting cleaned up properly. Contributed by Hitesh Shah.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1239443 13f79535-47bb-0310-9956-ffa450edef68

hadoop-mapreduce-project/CHANGES.txt

@@ -657,6 +657,9 @@ Release 0.23.1 - Unreleased
     MAPREDUCE-3744. Fix the yarn logs command line. Improve error messages for
     mapred job -logs. (Jason Lowe via sseth)
 
+    MAPREDUCE-3780. Fixed a bug where applications killed before getting
+    activated were not getting cleaned up properly. (Hitesh Shah via acmurthy)
+
 Release 0.23.0 - 2011-11-01 
 
   INCOMPATIBLE CHANGES

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

@@ -635,7 +635,7 @@ public class LeafQueue implements CSQueue {
         user.activateApplication();
         activeApplications.add(application);
         i.remove();
-        LOG.info("Application " + application.getApplicationId().getId() + 
+        LOG.info("Application " + application.getApplicationId() +
             " from user: " + application.getUser() + 
             " activated in queue: " + getQueueName());
       }
@@ -673,10 +673,13 @@ public class LeafQueue implements CSQueue {
   }
 
   public synchronized void removeApplication(SchedulerApp application, User user) {
-    activeApplications.remove(application);
+    boolean wasActive = activeApplications.remove(application);
+    if (!wasActive) {
+      pendingApplications.remove(application);
+    }
     applicationsMap.remove(application.getApplicationAttemptId());
 
-    user.finishApplication();
+    user.finishApplication(wasActive);
     if (user.getTotalApplications() == 0) {
       users.remove(application.getUser());
     }
@@ -1415,9 +1418,14 @@ public class LeafQueue implements CSQueue {
       ++activeApplications;
     }
 
-    public synchronized void finishApplication() {
+    public synchronized void finishApplication(boolean wasActive) {
+      if (wasActive) {
         --activeApplications;
       }
+      else {
+        --pendingApplications;
+      }
+    }
 
     public synchronized void assignContainer(Resource resource) {
       Resources.addTo(consumed, resource);

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

@@ -256,6 +256,87 @@ public class TestApplicationLimits {
     assertEquals(0, queue.getNumPendingApplications(user_1));
   }
 
+  @Test
+  public void testActiveLimitsWithKilledApps() throws Exception {
+    final String user_0 = "user_0";
+
+    int APPLICATION_ID = 0;
+
+    // set max active to 2
+    doReturn(2).when(queue).getMaximumActiveApplications();
+
+    // Submit first application
+    SchedulerApp app_0 = getMockApplication(APPLICATION_ID++, user_0);
+    queue.submitApplication(app_0, user_0, A);
+    assertEquals(1, queue.getNumActiveApplications());
+    assertEquals(0, queue.getNumPendingApplications());
+    assertEquals(1, queue.getNumActiveApplications(user_0));
+    assertEquals(0, queue.getNumPendingApplications(user_0));
+    assertTrue(queue.activeApplications.contains(app_0));
+
+    // Submit second application
+    SchedulerApp app_1 = getMockApplication(APPLICATION_ID++, user_0);
+    queue.submitApplication(app_1, user_0, A);
+    assertEquals(2, queue.getNumActiveApplications());
+    assertEquals(0, queue.getNumPendingApplications());
+    assertEquals(2, queue.getNumActiveApplications(user_0));
+    assertEquals(0, queue.getNumPendingApplications(user_0));
+    assertTrue(queue.activeApplications.contains(app_1));
+
+    // Submit third application, should remain pending
+    SchedulerApp app_2 = getMockApplication(APPLICATION_ID++, user_0);
+    queue.submitApplication(app_2, user_0, A);
+    assertEquals(2, queue.getNumActiveApplications());
+    assertEquals(1, queue.getNumPendingApplications());
+    assertEquals(2, queue.getNumActiveApplications(user_0));
+    assertEquals(1, queue.getNumPendingApplications(user_0));
+    assertTrue(queue.pendingApplications.contains(app_2));
+
+    // Submit fourth application, should remain pending
+    SchedulerApp app_3 = getMockApplication(APPLICATION_ID++, user_0);
+    queue.submitApplication(app_3, user_0, A);
+    assertEquals(2, queue.getNumActiveApplications());
+    assertEquals(2, queue.getNumPendingApplications());
+    assertEquals(2, queue.getNumActiveApplications(user_0));
+    assertEquals(2, queue.getNumPendingApplications(user_0));
+    assertTrue(queue.pendingApplications.contains(app_3));
+
+    // Kill 3rd pending application
+    queue.finishApplication(app_2, A);
+    assertEquals(2, queue.getNumActiveApplications());
+    assertEquals(1, queue.getNumPendingApplications());
+    assertEquals(2, queue.getNumActiveApplications(user_0));
+    assertEquals(1, queue.getNumPendingApplications(user_0));
+    assertFalse(queue.pendingApplications.contains(app_2));
+    assertFalse(queue.activeApplications.contains(app_2));
+
+    // Finish 1st application, app_3 should become active
+    queue.finishApplication(app_0, A);
+    assertEquals(2, queue.getNumActiveApplications());
+    assertEquals(0, queue.getNumPendingApplications());
+    assertEquals(2, queue.getNumActiveApplications(user_0));
+    assertEquals(0, queue.getNumPendingApplications(user_0));
+    assertTrue(queue.activeApplications.contains(app_3));
+    assertFalse(queue.pendingApplications.contains(app_3));
+    assertFalse(queue.activeApplications.contains(app_0));
+
+    // Finish 2nd application
+    queue.finishApplication(app_1, A);
+    assertEquals(1, queue.getNumActiveApplications());
+    assertEquals(0, queue.getNumPendingApplications());
+    assertEquals(1, queue.getNumActiveApplications(user_0));
+    assertEquals(0, queue.getNumPendingApplications(user_0));
+    assertFalse(queue.activeApplications.contains(app_1));
+
+    // Finish 4th application
+    queue.finishApplication(app_3, A);
+    assertEquals(0, queue.getNumActiveApplications());
+    assertEquals(0, queue.getNumPendingApplications());
+    assertEquals(0, queue.getNumActiveApplications(user_0));
+    assertEquals(0, queue.getNumPendingApplications(user_0));
+    assertFalse(queue.activeApplications.contains(app_3));
+  }
+
   @Test
   public void testHeadroom() throws Exception {
     CapacitySchedulerConfiguration csConf =