Making Sharad's new TestFifoScheduler run to successful completion.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/MR-279@1153442 13f79535-47bb-0310-9956-ffa450edef68

mapreduce/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java

@@ -223,7 +223,7 @@ public class TestNodeStatusUpdater {
         + nm.getServiceState());
 
     int waitCount = 0;
-    while (nm.getServiceState() == STATE.INITED || waitCount++ != 20) {
+    while (nm.getServiceState() == STATE.INITED && waitCount++ != 20) {
       LOG.info("Waiting for NM to start..");
       Thread.sleep(1000);
     }

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java

@@ -324,12 +324,22 @@ public class RMAppAttemptImpl implements RMAppAttempt {
 
   @Override
   public List<Container> getJustFinishedContainers() {
-    return this.justFinishedContainers;
+    this.readLock.lock();
+    try {
+      return this.justFinishedContainers;
+    } finally {
+      this.readLock.unlock();
+    }
   }
 
   @Override
   public List<Container> getNewlyAllocatedContainers() {
-    return this.newlyAllocatedContainers;
+    this.readLock.lock();
+    try {
+      return this.newlyAllocatedContainers;
+    } finally {
+      this.readLock.unlock();
+    }
   }
 
   @Override

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/RMContainerImpl.java

@@ -48,6 +48,8 @@ public class RMContainerImpl implements RMContainer {
     // Transitions from ACQUIRED state
     .addTransition(RMContainerState.ACQUIRED, RMContainerState.RUNNING,
         RMContainerEventType.LAUNCHED, new LaunchedTransition())
+    .addTransition(RMContainerState.ACQUIRED, RMContainerState.COMPLETED,
+        RMContainerEventType.FINISHED, new ContainerFinishedAtAcquiredState())        
     .addTransition(RMContainerState.ACQUIRED, RMContainerState.RELEASED,
         RMContainerEventType.RELEASED, new KillTransition())
     .addTransition(RMContainerState.ACQUIRED, RMContainerState.EXPIRED,
@@ -216,6 +218,19 @@ public class RMContainerImpl implements RMContainer {
     }
   }
 
+  private static final class ContainerFinishedAtAcquiredState extends
+      FinishedTransition {
+    @Override
+    public void transition(RMContainerImpl container, RMContainerEvent event) {
+
+      // Unregister from containerAllocationExpirer.
+      container.containerAllocationExpirer.unregister(container.containerId);
+
+      // Inform AppAttempt, scheduler etc.
+      super.transition(container, event);
+    }
+  }
+
   private static final class KillTransition extends FinishedTransition {
 
     @Override

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java

@@ -388,8 +388,8 @@ public class RMNodeImpl implements RMNode, EventHandler<RMNodeEvent> {
 
           // Process running containers
           ContainerId containerId = remoteContainer.getId();
-          if (remoteContainer.getContainerStatus().getState() == ContainerState.RUNNING
-              || remoteContainer.getContainerStatus().getState() == ContainerState.INITIALIZING) {
+          if (remoteContainer.getState() == ContainerState.RUNNING
+              || remoteContainer.getState() == ContainerState.INITIALIZING) {
             if (!rmNode.launchedContainers.containsKey(containerId)) {
               // Just launched container. RM knows about it the first time.
               rmNode.launchedContainers.put(containerId, remoteContainer);

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeStatusEvent.java

@@ -1,6 +1,5 @@
 package org.apache.hadoop.yarn.server.resourcemanager.rmnode;
 
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java

@@ -36,7 +36,8 @@ public class MockNM {
 
   public void containerStatus(Container container) throws Exception {
     Map<ApplicationId, List<Container>> conts = new HashMap<ApplicationId, List<Container>>();
-    conts.put(container.getId().getAppId(), Arrays.asList(new Container[]{}));
+    conts.put(container.getId().getAppId(), Arrays
+        .asList(new Container[] { container }));
     nodeHeartbeat(conts, true);
   }
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRM.java

@@ -8,7 +8,6 @@ import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationIdRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationIdResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
-import org.apache.hadoop.yarn.api.records.AMResponse;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestFifoScheduler.java

@@ -55,83 +55,92 @@ public class TestFifoScheduler {
   public void tearDown() throws Exception {
   }
   
-//  @Test
+  @Test
   public void test() throws Exception {
     Logger rootLogger = LogManager.getRootLogger();
     rootLogger.setLevel(Level.DEBUG);
     MockRM rm = new MockRM();
     rm.start();
-    int GB = 1000;
-    MockNM nm1 = rm.registerNode("h1:1234", 6*GB);
-    MockNM nm2 = rm.registerNode("h2:5678", 4*GB);
-    
-    RMApp app1 = rm.submitApp(2000);
-    //kick the scheduling, 2 GB given to AM1, remaining 4GB
+    int GB = 1024;
+    MockNM nm1 = rm.registerNode("h1:1234", 6 * GB);
+    MockNM nm2 = rm.registerNode("h2:5678", 4 * GB);
+
+    RMApp app1 = rm.submitApp(2048);
+    // kick the scheduling, 2 GB given to AM1, remaining 4GB
     nm1.nodeHeartbeat(true);
     RMAppAttempt attempt1 = app1.getCurrentAppAttempt();
     MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId());
     am1.registerAppAttempt();
-    Assert.assertEquals(2*GB, rm.getResourceScheduler().getUsedResource(nm1.getNodeId()).getMemory());
-    
-    
+    Assert.assertEquals(2 * GB, rm.getResourceScheduler().getUsedResource(
+        nm1.getNodeId()).getMemory());
 
-    RMApp app2 = rm.submitApp(2000);
-    //kick the scheduling, 2GB given to AM, remaining 2 GB
+    RMApp app2 = rm.submitApp(2048);
+    // kick the scheduling, 2GB given to AM, remaining 2 GB
     nm2.nodeHeartbeat(true);
     RMAppAttempt attempt2 = app2.getCurrentAppAttempt();
     MockAM am2 = rm.sendAMLaunched(attempt2.getAppAttemptId());
     am2.registerAppAttempt();
-    Assert.assertEquals(2*GB, rm.getResourceScheduler().getUsedResource(nm2.getNodeId()).getMemory());
-    
-  //add request for containers
-    am1.addRequests(new String[]{"h1", "h2"}, GB, 1, 1);
-    am1.schedule(); //send the request
-    //add request for containers
-    am2.addRequests(new String[]{"h1", "h2"}, 3*GB, 0, 1);
-    am2.schedule(); //send the request
-
-    //kick the scheduler, 1 GB and 3 GB given to AM1 and AM2, remaining 0
+    Assert.assertEquals(2 * GB, rm.getResourceScheduler().getUsedResource(
+        nm2.getNodeId()).getMemory());
+
+    // add request for containers
+    am1.addRequests(new String[] { "h1", "h2" }, GB, 1, 1);
+    am1.schedule(); // send the request
+    // add request for containers
+    am2.addRequests(new String[] { "h1", "h2" }, 3 * GB, 0, 1);
+    am2.schedule(); // send the request
+
+    // kick the scheduler, 1 GB and 3 GB given to AM1 and AM2, remaining 0
     nm1.nodeHeartbeat(true);
-    while(attempt1.getNewlyAllocatedContainers().size() < 1) {
+    while (attempt1.getNewlyAllocatedContainers().size() < 1) {
       LOG.info("Waiting for containers to be created for app 1...");
       Thread.sleep(1000);
     }
-    while(attempt2.getNewlyAllocatedContainers().size() < 1) {
+    while (attempt2.getNewlyAllocatedContainers().size() < 1) {
       LOG.info("Waiting for containers to be created for app 2...");
       Thread.sleep(1000);
     }
-    //kick the scheduler, nothing given remaining 2 GB.
+    // kick the scheduler, nothing given remaining 2 GB.
     nm2.nodeHeartbeat(true);
-    
-    AMResponse resp1 = am1.schedule(); //get allocations
+
+    AMResponse resp1 = am1.schedule(); // get allocations
     List<Container> allocated1 = resp1.getNewContainerList();
     Assert.assertEquals(1, allocated1.size());
-    Assert.assertEquals(1*GB, allocated1.get(0).getResource().getMemory());
+    Assert.assertEquals(1 * GB, allocated1.get(0).getResource().getMemory());
     Assert.assertEquals(nm1.getNodeId(), allocated1.get(0).getNodeId());
 
-    AMResponse resp2 = am2.schedule(); //get allocations
+    AMResponse resp2 = am2.schedule(); // get allocations
     List<Container> allocated2 = resp2.getNewContainerList();
     Assert.assertEquals(1, allocated2.size());
-    Assert.assertEquals(3*GB, allocated2.get(0).getResource().getMemory());
+    Assert.assertEquals(3 * GB, allocated2.get(0).getResource().getMemory());
     Assert.assertEquals(nm1.getNodeId(), allocated2.get(0).getNodeId());
-    
-    Assert.assertEquals(0, rm.getResourceScheduler().getAvailableResource(nm1.getNodeId()).getMemory());
-    Assert.assertEquals(2*GB, rm.getResourceScheduler().getAvailableResource(nm2.getNodeId()).getMemory());
-    
-    Assert.assertEquals(6*GB, rm.getResourceScheduler().getUsedResource(nm1.getNodeId()).getMemory());
-    Assert.assertEquals(2*GB, rm.getResourceScheduler().getUsedResource(nm2.getNodeId()).getMemory());
-    
+
+    Assert.assertEquals(0, rm.getResourceScheduler().getAvailableResource(
+        nm1.getNodeId()).getMemory());
+    Assert.assertEquals(2 * GB, rm.getResourceScheduler()
+        .getAvailableResource(nm2.getNodeId()).getMemory());
+
+    Assert.assertEquals(6 * GB, rm.getResourceScheduler().getUsedResource(
+        nm1.getNodeId()).getMemory());
+    Assert.assertEquals(2 * GB, rm.getResourceScheduler().getUsedResource(
+        nm2.getNodeId()).getMemory());
+
     Container c1 = allocated1.get(0);
     Assert.assertEquals(GB, c1.getResource().getMemory());
     c1.setState(ContainerState.COMPLETE);
     nm1.containerStatus(c1);
-    while(attempt1.getJustFinishedContainers().size() < 1) {
-      LOG.info("Waiting for containers to be finished for app 1...");
+    int waitCount = 0;
+    while (attempt1.getJustFinishedContainers().size() < 1
+        && waitCount++ != 20) {
+      LOG.info("Waiting for containers to be finished for app 1... Tried "
+          + waitCount + " times already..");
       Thread.sleep(1000);
     }
+    Assert.assertEquals(1, attempt1.getJustFinishedContainers().size());
     Assert.assertEquals(1, am1.schedule().getFinishedContainerList().size());
-    Assert.assertEquals(5*GB, rm.getResourceScheduler().getUsedResource(nm1.getNodeId()).getMemory());
-    
+    Assert.assertEquals(5 * GB, rm.getResourceScheduler().getUsedResource(
+        nm1.getNodeId()).getMemory());
+
     rm.stop();
   }
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/TestFifoScheduler.java

@@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo;
 import java.io.IOException;
 
 import junit.framework.Assert;
-import junit.framework.TestCase;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -38,7 +37,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Test;
 
 public class TestFifoScheduler {
   private static final Log LOG = LogFactory.getLog(TestFifoScheduler.class);

mapreduce/yarn/yarn-server/yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/TestContainerTokenSecretManager.java

@@ -169,8 +169,8 @@ public class TestContainerTokenSecretManager {
     RMAppAttempt appAttempt = app == null ? null : app.getCurrentAppAttempt();
     RMAppAttemptState state = appAttempt == null ? null : appAttempt
         .getAppAttemptState();
-    while (app == null || appAttempt == null || state == null
-        || !state.equals(RMAppAttemptState.LAUNCHED) || waitCounter++ != 20) {
+    while ((app == null || appAttempt == null || state == null
+        || !state.equals(RMAppAttemptState.LAUNCHED)) && waitCounter++ != 20) {
       LOG.info("Waiting for applicationAttempt to be created.. ");
       Thread.sleep(1000);
       app = resourceManager.getRMContext().getRMApps().get(appID);