@Test(timeout = 600000)
public void testContainerPromoteAndDemoteBeforeContainerStart() throws Exception {

  HashMap<NodeId, MockNM> nodes = new HashMap<>();
  MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService());
  nodes.put(nm1.getNodeId(), nm1);
  MockNM nm2 = new MockNM("h1:4321", 4096, rm.getResourceTrackerService());
  nodes.put(nm2.getNodeId(), nm2);
  MockNM nm3 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService());
  nodes.put(nm3.getNodeId(), nm3);
  MockNM nm4 = new MockNM("h2:4321", 4096, rm.getResourceTrackerService());
  nodes.put(nm4.getNodeId(), nm4);
  nm1.registerNode();
  nm2.registerNode();
  nm3.registerNode();
  nm4.registerNode();
  nm1.nodeHeartbeat(oppContainersStatus, true);
  nm2.nodeHeartbeat(oppContainersStatus, true);
  nm3.nodeHeartbeat(oppContainersStatus, true);
  nm4.nodeHeartbeat(oppContainersStatus, true);
  OpportunisticContainerAllocatorAMService amservice =
      (OpportunisticContainerAllocatorAMService) rm
          .getApplicationMasterService();
  MockRMAppSubmissionData data =
      MockRMAppSubmissionData.Builder.createWithMemory(1 * GB, rm)
          .withAppName("app")
          .withUser("user")
          .withAcls(null)
          .withQueue("default")
          .withUnmanagedAM(false)
          .build();
  RMApp app1 = MockRMAppSubmitter.submit(rm, data);
  MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2);
  ResourceScheduler scheduler = rm.getResourceScheduler();
  // All nodes 1 - 4 will be applicable for scheduling.
  nm1.nodeHeartbeat(oppContainersStatus, true);
  nm2.nodeHeartbeat(oppContainersStatus, true);
  nm3.nodeHeartbeat(oppContainersStatus, true);
  nm4.nodeHeartbeat(oppContainersStatus, true);
  GenericTestUtils.waitFor(() ->
      amservice.getLeastLoadedNodes().size() == 4, 10, 10 * 100);
  QueueMetrics metrics = ((CapacityScheduler) scheduler).getRootQueue()
      .getMetrics();
  // Verify Metrics
  verifyMetrics(metrics, 15360, 15, 1024, 1, 1);
  AllocateResponse allocateResponse = am1.allocate(
      Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1),
          "*", Resources.createResource(1 * GB), 2, true, null,
          ExecutionTypeRequest.newInstance(
              ExecutionType.OPPORTUNISTIC, true))),
      null);
  List<Container> allocatedContainers = allocateResponse
      .getAllocatedContainers();
  Assert.assertEquals(2, allocatedContainers.size());
  Container container = allocatedContainers.get(0);
  MockNM allocNode = nodes.get(container.getNodeId());
  MockNM sameHostDiffNode = null;
  for (NodeId n : nodes.keySet()) {
    if (n.getHost().equals(allocNode.getNodeId().getHost()) &&
        n.getPort() != allocNode.getNodeId().getPort()) {
      sameHostDiffNode = nodes.get(n);
    }
  }
  // Verify Metrics After OPP allocation (Nothing should change)
  verifyMetrics(metrics, 15360, 15, 1024, 1, 1);
  am1.sendContainerUpdateRequest(
      Arrays.asList(UpdateContainerRequest.newInstance(0,
          container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE,
          null, ExecutionType.GUARANTEED)));
  // Node on same host should not result in allocation
  sameHostDiffNode.nodeHeartbeat(oppContainersStatus, true);
  rm.drainEvents();
  allocateResponse =  am1.allocate(new ArrayList<>(), new ArrayList<>());
  Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size());
  // Wait for scheduler to process all events
  dispatcher.waitForEventThreadToWait();
  rm.drainEvents();
  // Verify Metrics After OPP allocation (Nothing should change again)
  verifyMetrics(metrics, 15360, 15, 1024, 1, 1);
  // Send Promotion req again... this should result in update error
  allocateResponse = am1.sendContainerUpdateRequest(
      Arrays.asList(UpdateContainerRequest.newInstance(0,
          container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE,
          null, ExecutionType.GUARANTEED)));
  Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size());
  Assert.assertEquals(1, allocateResponse.getUpdateErrors().size());
  Assert.assertEquals("UPDATE_OUTSTANDING_ERROR",
      allocateResponse.getUpdateErrors().get(0).getReason());
  Assert.assertEquals(container.getId(),
      allocateResponse.getUpdateErrors().get(0)
          .getUpdateContainerRequest().getContainerId());
  // Send Promotion req again with incorrect version...
  // this should also result in update error
  allocateResponse = am1.sendContainerUpdateRequest(
      Arrays.asList(UpdateContainerRequest.newInstance(1,
          container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE,
          null, ExecutionType.GUARANTEED)));
  Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size());
  Assert.assertEquals(1, allocateResponse.getUpdateErrors().size());
  Assert.assertEquals("INCORRECT_CONTAINER_VERSION_ERROR",
      allocateResponse.getUpdateErrors().get(0).getReason());
  Assert.assertEquals(0,
      allocateResponse.getUpdateErrors().get(0)
          .getCurrentContainerVersion());
  Assert.assertEquals(container.getId(),
      allocateResponse.getUpdateErrors().get(0)
          .getUpdateContainerRequest().getContainerId());
  // Ensure after correct node heartbeats, we should get the allocation
  allocNode.nodeHeartbeat(oppContainersStatus, true);
  rm.drainEvents();
  allocateResponse =  am1.allocate(new ArrayList<>(), new ArrayList<>());
  Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size());
  Container uc =
      allocateResponse.getUpdatedContainers().get(0).getContainer();
  Assert.assertEquals(ExecutionType.GUARANTEED, uc.getExecutionType());
  Assert.assertEquals(uc.getId(), container.getId());
  Assert.assertEquals(uc.getVersion(), container.getVersion() + 1);
  // Verify Metrics After OPP allocation :
  // Allocated cores+mem should have increased, available should decrease
  verifyMetrics(metrics, 14336, 14, 2048, 2, 2);
  nm1.nodeHeartbeat(oppContainersStatus, true);
  nm2.nodeHeartbeat(oppContainersStatus, true);
  nm3.nodeHeartbeat(oppContainersStatus, true);
  nm4.nodeHeartbeat(oppContainersStatus, true);
  rm.drainEvents();
  // Verify that the container is still in ACQUIRED state wrt the RM.
  RMContainer rmContainer = ((CapacityScheduler) scheduler)
      .getApplicationAttempt(
      uc.getId().getApplicationAttemptId()).getRMContainer(uc.getId());
  Assert.assertEquals(RMContainerState.ACQUIRED, rmContainer.getState());
  // Now demote the container back..
  allocateResponse = am1.sendContainerUpdateRequest(
      Arrays.asList(UpdateContainerRequest.newInstance(uc.getVersion(),
          uc.getId(), ContainerUpdateType.DEMOTE_EXECUTION_TYPE,
          null, ExecutionType.OPPORTUNISTIC)));
  // This should happen in the same heartbeat..
  Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size());
  uc = allocateResponse.getUpdatedContainers().get(0).getContainer();
  Assert.assertEquals(ExecutionType.OPPORTUNISTIC, uc.getExecutionType());
  Assert.assertEquals(uc.getId(), container.getId());
  Assert.assertEquals(uc.getVersion(), container.getVersion() + 2);
  // Wait for scheduler to finish processing events
  dispatcher.waitForEventThreadToWait();
  rm.drainEvents();
  // Verify Metrics After OPP allocation :
  // Everything should have reverted to what it was
  verifyMetrics(metrics, 15360, 15, 1024, 1, 1);
}