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

  rm.stop();
  createAndStartRMWithAutoUpdateContainer();
  MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService());
  nm1.registerNode();
  nm1.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, nm1);
  ResourceScheduler scheduler = rm.getResourceScheduler();
  RMNode rmNode1 = rm.getRMContext().getRMNodes().get(nm1.getNodeId());
  nm1.nodeHeartbeat(oppContainersStatus, true);
  GenericTestUtils.waitFor(() ->
      amservice.getLeastLoadedNodes().size() == 1, 10, 10 * 100);
  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();
  allocatedContainers.addAll(
      am1.allocate(null, null).getAllocatedContainers());
  Assert.assertEquals(2, allocatedContainers.size());
  Container container = allocatedContainers.get(0);
  // Start Container in NM
  nm1.nodeHeartbeat(Arrays.asList(ContainerStatus
      .newInstance(container.getId(), ExecutionType.OPPORTUNISTIC,
          ContainerState.RUNNING, "", 0)), true);
  rm.drainEvents();
  // Verify that container is actually running wrt the RM..
  RMContainer rmContainer = ((CapacityScheduler) scheduler)
      .getApplicationAttempt(container.getId().getApplicationAttemptId())
      .getRMContainer(container.getId());
  Assert.assertEquals(RMContainerState.RUNNING, rmContainer.getState());
  // Send Promotion req... this should result in update error
  // Since the container doesn't exist anymore..
  allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList(
      UpdateContainerRequest.newInstance(0, container.getId(),
          ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null,
          ExecutionType.GUARANTEED)));
  nm1.nodeHeartbeat(Arrays.asList(ContainerStatus
      .newInstance(container.getId(), ExecutionType.OPPORTUNISTIC,
          ContainerState.RUNNING, "", 0)), true);
  rm.drainEvents();
  // Get the update response on next allocate
  allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>());
  // Check the update response from YARNRM
  Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size());
  UpdatedContainer uc = allocateResponse.getUpdatedContainers().get(0);
  Assert.assertEquals(container.getId(), uc.getContainer().getId());
  Assert.assertEquals(ExecutionType.GUARANTEED,
      uc.getContainer().getExecutionType());
  // Check that the container is updated in NM through NM heartbeat response
  NodeHeartbeatResponse response = nm1.nodeHeartbeat(true);
  Assert.assertEquals(1, response.getContainersToUpdate().size());
  Container containersFromNM = response.getContainersToUpdate().get(0);
  Assert.assertEquals(container.getId(), containersFromNM.getId());
  Assert.assertEquals(ExecutionType.GUARANTEED,
      containersFromNM.getExecutionType());
  //Increase resources
  allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList(
      UpdateContainerRequest.newInstance(1, container.getId(),
          ContainerUpdateType.INCREASE_RESOURCE,
          Resources.createResource(2 * GB, 1), null)));
  response = nm1.nodeHeartbeat(Arrays.asList(ContainerStatus
      .newInstance(container.getId(), ExecutionType.GUARANTEED,
          ContainerState.RUNNING, "", 0)), true);
  rm.drainEvents();
  if (allocateResponse.getUpdatedContainers().size() == 0) {
    allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>());
  }
  Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size());
  uc = allocateResponse.getUpdatedContainers().get(0);
  Assert.assertEquals(container.getId(), uc.getContainer().getId());
  Assert.assertEquals(Resource.newInstance(2 * GB, 1),
      uc.getContainer().getResource());
  rm.drainEvents();
  // Check that the container resources are increased in
  // NM through NM heartbeat response
  if (response.getContainersToUpdate().size() == 0) {
    response = nm1.nodeHeartbeat(true);
  }
  Assert.assertEquals(1, response.getContainersToUpdate().size());
  Assert.assertEquals(Resource.newInstance(2 * GB, 1),
      response.getContainersToUpdate().get(0).getResource());
  //Decrease resources
  allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList(
      UpdateContainerRequest.newInstance(2, container.getId(),
          ContainerUpdateType.DECREASE_RESOURCE,
          Resources.createResource(1 * GB, 1), null)));
  Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size());
  rm.drainEvents();
  // Check that the container resources are decreased
  // in NM through NM heartbeat response
  response = nm1.nodeHeartbeat(true);
  Assert.assertEquals(1, response.getContainersToUpdate().size());
  Assert.assertEquals(Resource.newInstance(1 * GB, 1),
      response.getContainersToUpdate().get(0).getResource());
  nm1.nodeHeartbeat(oppContainersStatus, true);
  // DEMOTE the container
  allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList(
      UpdateContainerRequest.newInstance(3, container.getId(),
          ContainerUpdateType.DEMOTE_EXECUTION_TYPE, null,
          ExecutionType.OPPORTUNISTIC)));
  response = nm1.nodeHeartbeat(Arrays.asList(ContainerStatus
      .newInstance(container.getId(), ExecutionType.GUARANTEED,
          ContainerState.RUNNING, "", 0)), true);
  rm.drainEvents();
  if (allocateResponse.getUpdatedContainers().size() == 0) {
    // Get the update response on next allocate
    allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>());
  }
  // Check the update response from YARNRM
  Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size());
  uc = allocateResponse.getUpdatedContainers().get(0);
  Assert.assertEquals(ExecutionType.OPPORTUNISTIC,
      uc.getContainer().getExecutionType());
  // Check that the container is updated in NM through NM heartbeat response
  if (response.getContainersToUpdate().size() == 0) {
    response = nm1.nodeHeartbeat(oppContainersStatus, true);
  }
  Assert.assertEquals(1, response.getContainersToUpdate().size());
  Assert.assertEquals(ExecutionType.OPPORTUNISTIC,
      response.getContainersToUpdate().get(0).getExecutionType());
}