TestFifoScheduler and related fixes in app/container/scheduler.

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

mapreduce/yarn/yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerId.java

@@ -1,9 +1,11 @@
 package org.apache.hadoop.yarn.api.records;
 
 public interface ContainerId extends Comparable<ContainerId>{
+  public abstract ApplicationAttemptId getAppAttemptId();
   public abstract ApplicationId getAppId();
   public abstract int getId();
   
+  public abstract void setAppAttemptId(ApplicationAttemptId atId);
   public abstract void setAppId(ApplicationId appID);
   public abstract void setId(int id);
   

mapreduce/yarn/yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ContainerIdPBImpl.java

@@ -3,9 +3,11 @@ package org.apache.hadoop.yarn.api.records.impl.pb;
 
 import java.text.NumberFormat;
 
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ProtoBase;
+import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationAttemptIdProto;
 import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationIdProto;
 import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
 import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProtoOrBuilder;
@@ -18,6 +20,7 @@ public class ContainerIdPBImpl extends ProtoBase<ContainerIdProto> implements Co
   boolean viaProto = false;
   
   private ApplicationId applicationId = null;
+  private ApplicationAttemptId appAttemptId = null;
   protected static final NumberFormat idFormat = NumberFormat.getInstance();
   static {
     idFormat.setGroupingUsed(false);
@@ -51,6 +54,9 @@ public class ContainerIdPBImpl extends ProtoBase<ContainerIdProto> implements Co
     if (this.applicationId != null && !((ApplicationIdPBImpl)applicationId).getProto().equals(builder.getAppId())) {
       builder.setAppId(convertToProtoFormat(this.applicationId));
     }
+    if (this.appAttemptId != null && !((ApplicationAttemptIdPBImpl)appAttemptId).getProto().equals(builder.getAppAttemptId())) {
+      builder.setAppAttemptId(convertToProtoFormat(this.appAttemptId));
+    }
   }
 
   private void mergeLocalToProto() {
@@ -93,6 +99,19 @@ public class ContainerIdPBImpl extends ProtoBase<ContainerIdProto> implements Co
     return this.applicationId;
   }
 
+  @Override
+  public ApplicationAttemptId getAppAttemptId() {
+    ContainerIdProtoOrBuilder p = viaProto ? proto : builder;
+    if (this.appAttemptId != null) {
+      return this.appAttemptId;
+    }
+    if (!p.hasAppAttemptId()) {
+      return null;
+    }
+    this.appAttemptId = convertFromProtoFormat(p.getAppAttemptId());
+    return this.appAttemptId;
+  }
+
   @Override
   public void setAppId(ApplicationId appId) {
     maybeInitBuilder();
@@ -101,6 +120,22 @@ public class ContainerIdPBImpl extends ProtoBase<ContainerIdProto> implements Co
     this.applicationId = appId;
   }
 
+  @Override
+  public void setAppAttemptId(ApplicationAttemptId atId) {
+    maybeInitBuilder();
+    if (atId == null) 
+      builder.clearAppAttemptId();
+    this.appAttemptId = atId;
+  }
+
+  private ApplicationAttemptIdPBImpl convertFromProtoFormat(ApplicationAttemptIdProto p) {
+    return new ApplicationAttemptIdPBImpl(p);
+  }
+
+  private ApplicationAttemptIdProto convertToProtoFormat(ApplicationAttemptId t) {
+    return ((ApplicationAttemptIdPBImpl)t).getProto();
+  }
+
   private ApplicationIdPBImpl convertFromProtoFormat(ApplicationIdProto p) {
     return new ApplicationIdPBImpl(p);
   }

mapreduce/yarn/yarn-api/src/main/proto/yarn_protos.proto

@@ -22,7 +22,8 @@ message ApplicationAttemptIdProto {
 
 message ContainerIdProto {
   optional ApplicationIdProto app_id = 1;
-  optional int32 id = 2;
+  optional ApplicationAttemptIdProto app_attempt_id = 2;
+  optional int32 id = 3;
 }
 
 message ResourceProto {

mapreduce/yarn/yarn-common/src/main/java/org/apache/hadoop/yarn/util/BuilderUtils.java

@@ -21,6 +21,7 @@ package org.apache.hadoop.yarn.util;
 import java.net.URI;
 import java.util.Comparator;
 
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationState;
@@ -132,10 +133,11 @@ public class BuilderUtils {
   }
 
   public static ContainerId newContainerId(RecordFactory recordFactory,
-      ApplicationId applicationId,
+      ApplicationAttemptId appAttemptId,
       int containerId) {
     ContainerId id = recordFactory.newRecordInstance(ContainerId.class);
-    id.setAppId(applicationId);
+    id.setAppAttemptId(appAttemptId);
+    id.setAppId(appAttemptId.getApplicationId());
     id.setId(containerId);
     return id;
   }
@@ -152,11 +154,11 @@ public class BuilderUtils {
   }
 
   public static Container newContainer(RecordFactory recordFactory,
-      ApplicationId applicationId, int containerId, NodeId nodeId,
+      ApplicationAttemptId appAttemptId, int containerId, NodeId nodeId,
       String containerManagerAddress, String nodeHttpAddress,
       Resource resource) {
     ContainerId containerID =
-        newContainerId(recordFactory, applicationId, containerId);
+        newContainerId(recordFactory, appAttemptId, containerId);
     return newContainer(containerID, nodeId, containerManagerAddress,
         nodeHttpAddress, resource);
   }

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ams/ApplicationMasterService.java

@@ -239,10 +239,6 @@ AMRMProtocol, EventHandler<ApplicationMasterServiceEvent> {
       RMApp app = this.rmContext.getRMApps().get(appAttemptId.getApplicationId());
       RMAppAttempt appAttempt = app.getRMAppAttempt(appAttemptId);
 
-      // Get the list of finished containers.
-      List<Container> finishedContainers = appAttempt
-          .pullJustFinishedContainers();
-
       AMResponse response = recordFactory.newRecordInstance(AMResponse.class);
       response.addAllNewContainers(appAttempt.pullNewlyAllocatedContainers());
       response.addAllFinishedContainers(appAttempt

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

@@ -6,7 +6,6 @@ import java.util.Set;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
 import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.event.EventHandler;
 
@@ -34,6 +33,10 @@ public interface RMAppAttempt extends EventHandler<RMAppAttemptEvent>{
 
   List<Container> pullNewlyAllocatedContainers();
 
+  List<Container> getJustFinishedContainers();
+
+  List<Container> getNewlyAllocatedContainers();
+
   Container getMasterContainer();
 
   ApplicationSubmissionContext getSubmissionContext();

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

@@ -320,6 +320,16 @@ public class RMAppAttemptImpl implements RMAppAttempt {
     }
   }
 
+  @Override
+  public List<Container> getJustFinishedContainers() {
+    return this.justFinishedContainers;
+  }
+
+  @Override
+  public List<Container> getNewlyAllocatedContainers() {
+    return this.newlyAllocatedContainers;
+  }
+
   @Override
   public List<Container> pullJustFinishedContainers() {
     this.writeLock.lock();

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java

@@ -22,7 +22,7 @@ import java.io.IOException;
 import java.util.List;
 
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
 import org.apache.hadoop.yarn.api.records.Resource;
@@ -71,4 +71,8 @@ public interface YarnScheduler extends EventHandler<SchedulerEvent> {
 
   void allocate(ApplicationAttemptId appAttemptId, List<ResourceRequest> ask);
 
+  Resource getUsedResource(NodeId nodeId);
+
+  Resource getAvailableResource(NodeId nodeId);
+
 }

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

@@ -123,7 +123,7 @@ implements ResourceScheduler, CapacitySchedulerContext {
   private Resource maximumAllocation;
 
   private Map<ApplicationAttemptId, CSApp> applications = Collections
-      .synchronizedMap(new HashMap<ApplicationAttemptId, CSApp>());
+      .synchronizedMap(new TreeMap<ApplicationAttemptId, CSApp>());
 
   private boolean initialized = false;
 
@@ -151,6 +151,14 @@ implements ResourceScheduler, CapacitySchedulerContext {
     return maximumAllocation;
   }
 
+  public synchronized Resource getUsedResource(NodeId nodeId) {
+    return csNodes.get(nodeId).getUsedResource();
+  }
+
+  public synchronized Resource getAvailableResource(NodeId nodeId) {
+    return csNodes.get(nodeId).getAvailableResource();
+  }
+
   public synchronized int getNumClusterNodes() {
     return numNodeManagers;
   }
@@ -666,8 +674,7 @@ implements ResourceScheduler, CapacitySchedulerContext {
       Container container) {
     // Reap containers
     LOG.info("Application " + applicationId + " released container " + container);
-    // TODO:FIXMEVINODKV
-//    node.releaseContainer(container);
+    csNodes.get(container.getNodeId()).releaseContainer(container);
     return true;
   }
 

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

@@ -857,7 +857,7 @@ public class LeafQueue implements Queue {
         new ArrayList<Container>();
       Container container =
          BuilderUtils.newContainer(this.recordFactory,
-                    application.getApplicationId(),
+                    application.getApplicationAttemptId(),
                     application.getNewContainerId(),
                     node.getNodeID(), node.getNodeAddress(),
                     node.getHttpAddress(), capability);

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

@@ -55,7 +55,6 @@ import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
-import org.apache.hadoop.yarn.server.resourcemanager.recovery.Store.ApplicationInfo;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.Store.RMState;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
@@ -120,7 +119,7 @@ public class FifoScheduler implements ResourceScheduler {
   private Resource maximumAllocation;
 
   private Map<ApplicationAttemptId, SchedulerApp> applications
-      = new HashMap<ApplicationAttemptId, SchedulerApp>();
+      = new TreeMap<ApplicationAttemptId, SchedulerApp>();
 
   private static final String DEFAULT_QUEUE_NAME = "default";
   private final QueueMetrics metrics =
@@ -168,7 +167,15 @@ public class FifoScheduler implements ResourceScheduler {
       return Collections.singletonList(queueUserAclInfo);
     }
   };
-  
+
+  public synchronized Resource getUsedResource(NodeId nodeId) {
+    return nodes.get(nodeId).getUsedResource();
+  }
+
+  public synchronized Resource getAvailableResource(NodeId nodeId) {
+    return nodes.get(nodeId).getAvailableResource();
+  }
+
   @Override
   public Resource getMinimumResourceCapability() {
     return minimumAllocation;
@@ -471,10 +478,11 @@ public class FifoScheduler implements ResourceScheduler {
       for (int i=0; i < assignedContainers; ++i) {
         Container container =
             BuilderUtils.newContainer(recordFactory,
-                application.getApplicationId(),
+                application.getApplicationAttemptId(),
                 application.getNewContainerId(),
                 node.getRMNode().getNodeID(), node.getRMNode().getNodeAddress(),
                 node.getRMNode().getHttpAddress(), capability);
+        
         // If security is enabled, send the container-tokens too.
         if (UserGroupInformation.isSecurityEnabled()) {
           ContainerToken containerToken =
@@ -513,7 +521,7 @@ public class FifoScheduler implements ResourceScheduler {
   }
 
   private synchronized void applicationCompletedContainer(Container c) {
-      SchedulerApp app = applications.get(c.getId().getAppId());
+      SchedulerApp app = applications.get(c.getId().getAppAttemptId());
       /** this is possible, since an application can be removed from scheduler but
        * the nodemanger is just updating about a completed container.
        */
@@ -638,8 +646,7 @@ public class FifoScheduler implements ResourceScheduler {
     // Reap containers
     LOG.info("Application " + applicationId + " released container " +
         container.getId());
-    // TODO:FIXMEVINODKV
-//    node.releaseContainer(container);
+    nodes.get(container.getNodeId()).releaseContainer(container);
   }
 
   private synchronized void addAllocatedContainers(SchedulerNode node,

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

@@ -267,8 +267,11 @@ public class Application {
     // Get resources from the ResourceManager
     resourceManager.getResourceScheduler().allocate(applicationAttemptId,
         new ArrayList<ResourceRequest>(ask));
+    System.out.println("-=======" + applicationAttemptId);
+    System.out.println("----------" + resourceManager.getRMContext().getRMApps()
+        .get(applicationId).getRMAppAttempt(applicationAttemptId));
     
-    List<Container> containers = resourceManager.getRMContext().getRMApps()
+     List<Container> containers = resourceManager.getRMContext().getRMApps()
         .get(applicationId).getRMAppAttempt(applicationAttemptId)
         .pullNewlyAllocatedContainers();
 

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

@@ -1,7 +1,6 @@
 package org.apache.hadoop.yarn.server.resourcemanager;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 
 import junit.framework.Assert;
@@ -29,6 +28,9 @@ public class MockAM {
   private final ApplicationAttemptId attemptId;
   private final RMContext context;
   private final AMRMProtocol amRMProtocol;
+  
+  private final List<ResourceRequest> requests = new ArrayList<ResourceRequest>();
+  private final List<Container> releases = new ArrayList<Container>();
 
   MockAM(RMContext context, AMRMProtocol amRMProtocol, 
       ApplicationAttemptId attemptId) {
@@ -64,10 +66,19 @@ public class MockAM {
     amRMProtocol.registerApplicationMaster(req);
   }
 
+  public void addRequests(String[] hosts, int memory, int priority, 
+      int containers) throws Exception {
+    requests.addAll(createReq(hosts, memory, priority, containers));
+  }
+
+  public AMResponse schedule() throws Exception {
+    return allocate(releases, requests);
+  }
+
   public AMResponse allocate( 
       String host, int memory, int numContainers, 
       List<ContainerId> releases) throws Exception {
-    List reqs = createReq(host, memory, 1, numContainers);
+    List reqs = createReq(new String[]{host}, memory, 1, numContainers);
     List<Container> toRelease = new ArrayList<Container>();
     for (ContainerId id : releases) {
       Container cont = Records.newRecord(Container.class);
@@ -77,18 +88,26 @@ public class MockAM {
     return allocate(toRelease, reqs);
   }
 
-  private List<ResourceRequest> createReq(String host, int memory, int priority, 
+  public List<ResourceRequest> createReq(String[] hosts, int memory, int priority, 
       int containers) throws Exception {
-    ResourceRequest hostReq = createResourceReq(host, memory, priority, 
-        containers);
-    ResourceRequest rackReq = createResourceReq("default-rack", memory, 
-        priority, containers);
+    List<ResourceRequest> reqs = new ArrayList<ResourceRequest>();
+    for (String host : hosts) {
+      ResourceRequest hostReq = createResourceReq(host, memory, priority, 
+          containers);
+      reqs.add(hostReq);
+      ResourceRequest rackReq = createResourceReq("default-rack", memory, 
+          priority, containers);
+      reqs.add(rackReq);
+    }
+    
     ResourceRequest offRackReq = createResourceReq("*", memory, priority, 
         containers);
-    return Arrays.asList(new ResourceRequest[] {hostReq, rackReq, offRackReq});
+    reqs.add(offRackReq);
+    return reqs;
     
   }
-  private ResourceRequest createResourceReq(String resource, int memory, int priority, 
+
+  public ResourceRequest createResourceReq(String resource, int memory, int priority, 
       int containers) throws Exception {
     ResourceRequest req = Records.newRecord(ResourceRequest.class);
     req.setHostName(resource);

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

@@ -30,6 +30,10 @@ public class MockNM {
     this.resourceTracker = resourceTracker;
   }
 
+  public NodeId getNodeId() {
+    return nodeId;
+  }
+
   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[]{}));

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

@@ -94,6 +94,7 @@ public class NodeManager implements ContainerManager {
     RegisterNodeManagerRequest request = recordFactory
         .newRecordInstance(RegisterNodeManagerRequest.class);
     request.setHttpPort(httpPort);
+    request.setNodeId(this.nodeId);
     request.setResource(capability);
     RegistrationResponse response = resourceTrackerService
         .registerNodeManager(request).getRegistrationResponse();

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

@@ -0,0 +1,142 @@
+/**
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.hadoop.yarn.server.resourcemanager;
+
+import java.util.List;
+
+import junit.framework.Assert;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.AMResponse;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.Store;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.StoreFactory;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
+import org.apache.log4j.Level;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestFifoScheduler {
+  private static final Log LOG = LogFactory.getLog(TestFifoScheduler.class);
+  
+  private ResourceManager resourceManager = null;
+  
+  @Before
+  public void setUp() throws Exception {
+    Store store = StoreFactory.getStore(new Configuration());
+    resourceManager = new ResourceManager(store);
+    resourceManager.init(new Configuration());
+  }
+
+  @After
+  public void tearDown() throws Exception {
+  }
+  
+  @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
+    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());
+    
+    
+
+    RMApp app2 = rm.submitApp(2000);
+    //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
+    nm1.nodeHeartbeat(true);
+    while(attempt1.getNewlyAllocatedContainers().size() < 1) {
+      LOG.info("Waiting for containers to be created for app 1...");
+      Thread.sleep(1000);
+    }
+    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.
+    nm2.nodeHeartbeat(true);
+    
+    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(nm1.getNodeId(), allocated1.get(0).getNodeId());
+
+    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(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());
+    
+    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...");
+      Thread.sleep(1000);
+    }
+    Assert.assertEquals(1, am1.schedule().getFinishedContainerList().size());
+    Assert.assertEquals(5*GB, rm.getResourceScheduler().getUsedResource(nm1.getNodeId()).getMemory());
+    
+    rm.stop();
+  }
+
+  public static void main(String[] args) throws Exception {
+    TestFifoScheduler t = new TestFifoScheduler();
+    t.test();
+  }
+}

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

@@ -283,4 +283,10 @@ public class TestFifoScheduler extends TestCase {
     node.checkResourceUsage();
   }
 
+  public static void main(String[] arg) throws Exception {
+    TestFifoScheduler t = new TestFifoScheduler();
+    t.setUp();
+    t.testFifoScheduler();
+    t.tearDown();
+  }
 }