@SuppressWarnings("unchecked")
@Test
public void testHandleOpportunisticContainerStatus() throws Exception{

  final DrainDispatcher dispatcher = new DrainDispatcher();
  YarnConfiguration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
  conf.setBoolean(YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED,
      true);
  rm = new MockRM(conf){
    @Override
    protected Dispatcher createDispatcher() {
      return dispatcher;
    }
  };
  rm.start();
  MockRMAppSubmissionData data =
      MockRMAppSubmissionData.Builder.createWithMemory(1024, rm)
          .withUnmanagedAM(true)
          .build();
  RMApp app = MockRMAppSubmitter.submit(rm, data);
  ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt()
      .getAppAttemptId();
  ResourceTrackerService resourceTrackerService =
      rm.getResourceTrackerService();
  SchedulerApplicationAttempt applicationAttempt = null;
  while (applicationAttempt == null) {
    applicationAttempt =
        ((AbstractYarnScheduler)rm.getRMContext().getScheduler())
        .getApplicationAttempt(appAttemptId);
    Thread.sleep(100);
  }
  Resource currentConsumption = applicationAttempt.getCurrentConsumption();
  Assert.assertEquals(Resource.newInstance(0, 0), currentConsumption);
  Resource allocResources =
      applicationAttempt.getQueue().getMetrics().getAllocatedResources();
  Assert.assertEquals(Resource.newInstance(0, 0), allocResources);
  RegisterNodeManagerRequest req = Records.newRecord(
      RegisterNodeManagerRequest.class);
  NodeId nodeId = NodeId.newInstance("host2", 1234);
  Resource capability = BuilderUtils.newResource(1024, 1);
  NodeStatus mockNodeStatus = createMockNodeStatus();
  req.setResource(capability);
  req.setNodeId(nodeId);
  req.setHttpPort(1234);
  req.setNMVersion(YarnVersionInfo.getVersion());
  req.setNodeStatus(mockNodeStatus);
  ContainerId c1 = ContainerId.newContainerId(appAttemptId, 1);
  ContainerId c2 = ContainerId.newContainerId(appAttemptId, 2);
  ContainerId c3 = ContainerId.newContainerId(appAttemptId, 3);
  NMContainerStatus queuedOpp =
      NMContainerStatus.newInstance(c1, 1, ContainerState.RUNNING,
          Resource.newInstance(1024, 1), "Dummy Queued OC",
          ContainerExitStatus.INVALID, Priority.newInstance(5), 1234, "",
          ExecutionType.OPPORTUNISTIC, -1);
  NMContainerStatus runningOpp =
      NMContainerStatus.newInstance(c2, 1, ContainerState.RUNNING,
          Resource.newInstance(2048, 1), "Dummy Running OC",
          ContainerExitStatus.INVALID, Priority.newInstance(6), 1234, "",
          ExecutionType.OPPORTUNISTIC, -1);
  NMContainerStatus runningGuar =
      NMContainerStatus.newInstance(c3, 1, ContainerState.RUNNING,
          Resource.newInstance(2048, 1), "Dummy Running GC",
          ContainerExitStatus.INVALID, Priority.newInstance(6), 1234, "",
          ExecutionType.GUARANTEED, -1);
  req.setContainerStatuses(Arrays.asList(queuedOpp, runningOpp, runningGuar));
  // trying to register a invalid node.
  RegisterNodeManagerResponse response =
      resourceTrackerService.registerNodeManager(req);
  dispatcher.await();
  Thread.sleep(2000);
  dispatcher.await();
  Assert.assertEquals(NodeAction.NORMAL, response.getNodeAction());
  Collection<RMContainer> liveContainers = applicationAttempt
      .getLiveContainers();
  Assert.assertEquals(3, liveContainers.size());
  Iterator<RMContainer> iter = liveContainers.iterator();
  while (iter.hasNext()) {
    RMContainer rc = iter.next();
    Assert.assertEquals(
        rc.getContainerId().equals(c3) ?
            ExecutionType.GUARANTEED : ExecutionType.OPPORTUNISTIC,
        rc.getExecutionType());
  }
  // Should only include GUARANTEED resources
  currentConsumption = applicationAttempt.getCurrentConsumption();
  Assert.assertEquals(Resource.newInstance(2048, 1), currentConsumption);
  allocResources =
      applicationAttempt.getQueue().getMetrics().getAllocatedResources();
  Assert.assertEquals(Resource.newInstance(2048, 1), allocResources);
  SchedulerNode schedulerNode =
      rm.getRMContext().getScheduler().getSchedulerNode(nodeId);
  Assert.assertNotNull(schedulerNode);
  Resource nodeResources = schedulerNode.getAllocatedResource();
  Assert.assertEquals(Resource.newInstance(2048, 1), nodeResources);
}