@Test (timeout = 10000)
public void testContinuousScheduling() throws Exception {

  // set continuous scheduling enabled
  FairScheduler fs = new FairScheduler();
  Configuration conf = createConfiguration();
  conf.setBoolean(FairSchedulerConfiguration.CONTINUOUS_SCHEDULING_ENABLED,
          true);
  fs.reinitialize(conf, resourceManager.getRMContext());
  Assert.assertTrue("Continuous scheduling should be enabled.",
          fs.isContinuousSchedulingEnabled());
  // Add two nodes
  RMNode node1 =
          MockNodes.newNodeInfo(1, Resources.createResource(8 * 1024, 8), 1,
                  "127.0.0.1");
  NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
  fs.handle(nodeEvent1);
  RMNode node2 =
          MockNodes.newNodeInfo(1, Resources.createResource(8 * 1024, 8), 2,
                  "127.0.0.2");
  NodeAddedSchedulerEvent nodeEvent2 = new NodeAddedSchedulerEvent(node2);
  fs.handle(nodeEvent2);
  // available resource
  Assert.assertEquals(fs.getClusterCapacity().getMemory(), 16 * 1024);
  Assert.assertEquals(fs.getClusterCapacity().getVirtualCores(), 16);
  // send application request
  ApplicationAttemptId appAttemptId =
          createAppAttemptId(this.APP_ID++, this.ATTEMPT_ID++);
  fs.addApplication(appAttemptId.getApplicationId(), "queue11", "user11");
  fs.addApplicationAttempt(appAttemptId, false);
  List<ResourceRequest> ask = new ArrayList<ResourceRequest>();
  ResourceRequest request =
          createResourceRequest(1024, 1, ResourceRequest.ANY, 1, 1, true);
  ask.add(request);
  fs.allocate(appAttemptId, ask, new ArrayList<ContainerId>(), null, null);
  // waiting for continuous_scheduler_sleep_time
  // at least one pass
  Thread.sleep(fs.getConf().getContinuousSchedulingSleepMs() + 500);
  FSSchedulerApp app = fs.getSchedulerApp(appAttemptId);
  // Wait until app gets resources.
  while (app.getCurrentConsumption().equals(Resources.none())) { }
  // check consumption
  Assert.assertEquals(1024, app.getCurrentConsumption().getMemory());
  Assert.assertEquals(1, app.getCurrentConsumption().getVirtualCores());
  // another request
  request =
          createResourceRequest(1024, 1, ResourceRequest.ANY, 2, 1, true);
  ask.clear();
  ask.add(request);
  fs.allocate(appAttemptId, ask, new ArrayList<ContainerId>(), null, null);
  // Wait until app gets resources
  while (app.getCurrentConsumption()
          .equals(Resources.createResource(1024, 1))) { }
  Assert.assertEquals(2048, app.getCurrentConsumption().getMemory());
  Assert.assertEquals(2, app.getCurrentConsumption().getVirtualCores());
  // 2 containers should be assigned to 2 nodes
  Set<NodeId> nodes = new HashSet<NodeId>();
  Iterator<RMContainer> it = app.getLiveContainers().iterator();
  while (it.hasNext()) {
    nodes.add(it.next().getContainer().getNodeId());
  }
  Assert.assertEquals(2, nodes.size());
}