@Test
public void testCapacityScheduler() throws Exception {

  LOG.info("--- START: testCapacityScheduler ---");
      
  // Register node1
  String host_0 = "host_0";
  org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm_0 = 
    registerNode(host_0, 1234, 2345, NetworkTopology.DEFAULT_RACK, 
        Resources.createResource(4 * GB, 1));
  
  // Register node2
  String host_1 = "host_1";
  org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm_1 = 
    registerNode(host_1, 1234, 2345, NetworkTopology.DEFAULT_RACK, 
        Resources.createResource(2 * GB, 1));
  // ResourceRequest priorities
  Priority priority_0 = 
    org.apache.hadoop.yarn.server.resourcemanager.resource.Priority.create(0); 
  Priority priority_1 = 
    org.apache.hadoop.yarn.server.resourcemanager.resource.Priority.create(1);
  
  // Submit an application
  Application application_0 = new Application("user_0", "a1", resourceManager);
  application_0.submit();
  
  application_0.addNodeManager(host_0, 1234, nm_0);
  application_0.addNodeManager(host_1, 1234, nm_1);
  Resource capability_0_0 = Resources.createResource(1 * GB, 1);
  application_0.addResourceRequestSpec(priority_1, capability_0_0);
  
  Resource capability_0_1 = Resources.createResource(2 * GB, 1);
  application_0.addResourceRequestSpec(priority_0, capability_0_1);
  Task task_0_0 = new Task(application_0, priority_1, 
      new String[] {host_0, host_1});
  application_0.addTask(task_0_0);
     
  // Submit another application
  Application application_1 = new Application("user_1", "b2", resourceManager);
  application_1.submit();
  
  application_1.addNodeManager(host_0, 1234, nm_0);
  application_1.addNodeManager(host_1, 1234, nm_1);
  
  Resource capability_1_0 = Resources.createResource(3 * GB, 1);
  application_1.addResourceRequestSpec(priority_1, capability_1_0);
  
  Resource capability_1_1 = Resources.createResource(2 * GB, 1);
  application_1.addResourceRequestSpec(priority_0, capability_1_1);
  Task task_1_0 = new Task(application_1, priority_1, 
      new String[] {host_0, host_1});
  application_1.addTask(task_1_0);
      
  // Send resource requests to the scheduler
  application_0.schedule();
  application_1.schedule();
  // Send a heartbeat to kick the tires on the Scheduler
  LOG.info("Kick!");
  
  // task_0_0 and task_1_0 allocated, used=4G
  nodeUpdate(nm_0);
  
  // nothing allocated
  nodeUpdate(nm_1);
  
  // Get allocations from the scheduler
  application_0.schedule();     // task_0_0 
  checkApplicationResourceUsage(1 * GB, application_0);
  application_1.schedule();     // task_1_0
  checkApplicationResourceUsage(3 * GB, application_1);
  
  checkNodeResourceUsage(4*GB, nm_0);  // task_0_0 (1G) and task_1_0 (3G)
  checkNodeResourceUsage(0*GB, nm_1);  // no tasks, 2G available
  LOG.info("Adding new tasks...");
  
  Task task_1_1 = new Task(application_1, priority_0, 
      new String[] {ResourceRequest.ANY});
  application_1.addTask(task_1_1);
  application_1.schedule();
  Task task_0_1 = new Task(application_0, priority_0, 
      new String[] {host_0, host_1});
  application_0.addTask(task_0_1);
  application_0.schedule();
  // Send a heartbeat to kick the tires on the Scheduler
  LOG.info("Sending hb from " + nm_0.getHostName());
  // nothing new, used=4G
  nodeUpdate(nm_0);
  
  LOG.info("Sending hb from " + nm_1.getHostName());
  // task_0_1 is prefer as locality, used=2G
  nodeUpdate(nm_1);
  
  // Get allocations from the scheduler
  LOG.info("Trying to allocate...");
  application_0.schedule();
  checkApplicationResourceUsage(1 * GB, application_0);
  application_1.schedule();
  checkApplicationResourceUsage(5 * GB, application_1);
  
  nodeUpdate(nm_0);
  nodeUpdate(nm_1);
  
  checkNodeResourceUsage(4*GB, nm_0);
  checkNodeResourceUsage(2*GB, nm_1);
  LOG.info("--- END: testCapacityScheduler ---");
}