@Test
public void testResourceOverCommit() throws Exception {

  Configuration conf = new Configuration();
  conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
      ResourceScheduler.class);
  MockRM rm = new MockRM(conf);
  rm.start();
  ResourceScheduler scheduler = rm.getResourceScheduler();
  MockNM nm = rm.registerNode("127.0.0.1:1234", 4 * GB);
  NodeId nmId = nm.getNodeId();
  RMApp app = MockRMAppSubmitter.submitWithMemory(2048, rm);
  // kick the scheduling, 2 GB given to AM1, remaining 2GB on nm
  nm.nodeHeartbeat(true);
  RMAppAttempt attempt1 = app.getCurrentAppAttempt();
  MockAM am = rm.sendAMLaunched(attempt1.getAppAttemptId());
  am.registerAppAttempt();
  assertMemory(scheduler, nmId, 2 * GB, 2 * GB);
  // add request for 1 container of 2 GB
  am.addRequests(new String[] {"127.0.0.1", "127.0.0.2"}, 2 * GB, 1, 1);
  AllocateResponse alloc1Response = am.schedule(); // send the request
  // kick the scheduler, 2 GB given to AM1, resource remaining 0
  nm.nodeHeartbeat(true);
  while (alloc1Response.getAllocatedContainers().isEmpty()) {
    LOG.info("Waiting for containers to be created for app 1...");
    Thread.sleep(100);
    alloc1Response = am.schedule();
  }
  List<Container> allocated1 = alloc1Response.getAllocatedContainers();
  assertEquals(1, allocated1.size());
  Container c1 = allocated1.get(0);
  assertEquals(2 * GB, c1.getResource().getMemorySize());
  assertEquals(nmId, c1.getNodeId());
  // check node report, 4 GB used and 0 GB available
  assertMemory(scheduler, nmId, 4 * GB, 0);
  nm.nodeHeartbeat(true);
  assertEquals(4 * GB, nm.getCapability().getMemorySize());
  // update node resource to 2 GB, so resource is over-consumed
  updateNodeResource(rm, nmId, 2 * GB, 2, -1);
  // the used resource should still 4 GB and negative available resource
  waitMemory(scheduler, nmId, 4 * GB, -2 * GB, 200, 5 * 1000);
  // check that we did not get a preemption requests
  assertNoPreemption(am.schedule().getPreemptionMessage());
  // check that the NM got the updated resources
  nm.nodeHeartbeat(true);
  assertEquals(2 * GB, nm.getCapability().getMemorySize());
  // check container can complete successfully with resource over-commitment
  ContainerStatus containerStatus = BuilderUtils.newContainerStatus(
      c1.getId(), ContainerState.COMPLETE, "", 0, c1.getResource());
  nm.containerStatus(containerStatus);
  LOG.info("Waiting for containers to be finished for app 1...");
  GenericTestUtils.waitFor(
      () -> attempt1.getJustFinishedContainers().size() == 1, 100, 2000);
  assertEquals(1, am.schedule().getCompletedContainersStatuses().size());
  assertMemory(scheduler, nmId, 2 * GB, 0);
  // verify no NPE is trigger in schedule after resource is updated
  am.addRequests(new String[] {"127.0.0.1", "127.0.0.2"}, 3 * GB, 1, 1);
  AllocateResponse allocResponse2 = am.schedule();
  assertTrue("Shouldn't have enough resource to allocate containers",
      allocResponse2.getAllocatedContainers().isEmpty());
  // try 10 times as scheduling is an async process
  for (int i = 0; i < 10; i++) {
    Thread.sleep(100);
    allocResponse2 = am.schedule();
    assertTrue("Shouldn't have enough resource to allocate containers",
        allocResponse2.getAllocatedContainers().isEmpty());
  }
  // increase the resources again to 5 GB to schedule the 3GB container
  updateNodeResource(rm, nmId, 5 * GB, 2, -1);
  waitMemory(scheduler, nmId, 2 * GB, 3 * GB, 100, 5 * 1000);
  // kick the scheduling and check it took effect
  nm.nodeHeartbeat(true);
  while (allocResponse2.getAllocatedContainers().isEmpty()) {
    LOG.info("Waiting for containers to be created for app 1...");
    Thread.sleep(100);
    allocResponse2 = am.schedule();
  }
  assertEquals(1, allocResponse2.getAllocatedContainers().size());
  Container c2 = allocResponse2.getAllocatedContainers().get(0);
  assertEquals(3 * GB, c2.getResource().getMemorySize());
  assertEquals(nmId, c2.getNodeId());
  assertMemory(scheduler, nmId, 5 * GB, 0);
  // reduce the resources and trigger a preempt request to the AM for c2
  updateNodeResource(rm, nmId, 3 * GB, 2, 2 * 1000);
  waitMemory(scheduler, nmId, 5 * GB, -2 * GB, 200, 5 * 1000);
  PreemptionMessage preemptMsg = am.schedule().getPreemptionMessage();
  assertPreemption(c2.getId(), preemptMsg);
  // increasing the resources again, should stop killing the containers
  updateNodeResource(rm, nmId, 5 * GB, 2, -1);
  waitMemory(scheduler, nmId, 5 * GB, 0, 200, 5 * 1000);
  Thread.sleep(3 * 1000);
  assertMemory(scheduler, nmId, 5 * GB, 0);
  // reduce the resources again to trigger a preempt request to the AM for c2
  long t0 = Time.now();
  updateNodeResource(rm, nmId, 3 * GB, 2, 2 * 1000);
  waitMemory(scheduler, nmId, 5 * GB, -2 * GB, 200, 5 * 1000);
  preemptMsg = am.schedule().getPreemptionMessage();
  assertPreemption(c2.getId(), preemptMsg);
  // wait until the scheduler kills the container
  GenericTestUtils.waitFor(() -> {
    try {
      nm.nodeHeartbeat(true); // trigger preemption in the NM
    } catch (Exception e) {
      LOG.error("Cannot heartbeat", e);
    }
    SchedulerNodeReport report = scheduler.getNodeReport(nmId);
    return report.getAvailableResource().getMemorySize() > 0;
  }, 200, 5 * 1000);
  assertMemory(scheduler, nmId, 2 * GB, 1 * GB);
  List<ContainerStatus> completedContainers =
      am.schedule().getCompletedContainersStatuses();
  assertEquals(1, completedContainers.size());
  ContainerStatus c2status = completedContainers.get(0);
  assertContainerKilled(c2.getId(), c2status);
  assertTime(2000, Time.now() - t0);
  rm.stop();
}