@Test
public void testCompletedContainerEvent() {

  RMContainerAllocator allocator = new RMContainerAllocator(
      mock(ClientService.class), mock(AppContext.class),
      new NoopAMPreemptionPolicy());
  TaskAttemptId attemptId = MRBuilderUtils.newTaskAttemptId(
      MRBuilderUtils.newTaskId(
          MRBuilderUtils.newJobId(1, 1, 1), 1, TaskType.MAP), 1);
  ApplicationId applicationId = ApplicationId.newInstance(1, 1);
  ApplicationAttemptId applicationAttemptId =
      ApplicationAttemptId.newInstance(applicationId, 1);
  // ABORTED
  ContainerId containerId =
      ContainerId.newContainerId(applicationAttemptId, 1);
  ContainerStatus status = ContainerStatus.newInstance(
      containerId, ContainerState.RUNNING, "", 0);
  ContainerStatus abortedStatus = ContainerStatus.newInstance(
      containerId, ContainerState.RUNNING, "",
      ContainerExitStatus.ABORTED);
  TaskAttemptEvent event = allocator.createContainerFinishedEvent(status,
      attemptId);
  Assert.assertEquals(TaskAttemptEventType.TA_CONTAINER_COMPLETED,
      event.getType());
  TaskAttemptEvent abortedEvent = allocator.createContainerFinishedEvent(
      abortedStatus, attemptId);
  Assert.assertEquals(TaskAttemptEventType.TA_KILL, abortedEvent.getType());
  // PREEMPTED
  ContainerId containerId2 =
      ContainerId.newContainerId(applicationAttemptId, 2);
  ContainerStatus status2 = ContainerStatus.newInstance(containerId2,
      ContainerState.RUNNING, "", 0);
  ContainerStatus preemptedStatus = ContainerStatus.newInstance(containerId2,
      ContainerState.RUNNING, "", ContainerExitStatus.PREEMPTED);
  TaskAttemptEvent event2 = allocator.createContainerFinishedEvent(status2,
      attemptId);
  Assert.assertEquals(TaskAttemptEventType.TA_CONTAINER_COMPLETED,
      event2.getType());
  TaskAttemptEvent abortedEvent2 = allocator.createContainerFinishedEvent(
      preemptedStatus, attemptId);
  Assert.assertEquals(TaskAttemptEventType.TA_KILL, abortedEvent2.getType());
  // KILLED_BY_CONTAINER_SCHEDULER
  ContainerId containerId3 =
      ContainerId.newContainerId(applicationAttemptId, 3);
  ContainerStatus status3 = ContainerStatus.newInstance(containerId3,
      ContainerState.RUNNING, "", 0);
  ContainerStatus killedByContainerSchedulerStatus =
      ContainerStatus.newInstance(containerId3, ContainerState.RUNNING, "",
          ContainerExitStatus.KILLED_BY_CONTAINER_SCHEDULER);
  TaskAttemptEvent event3 = allocator.createContainerFinishedEvent(status3,
      attemptId);
  Assert.assertEquals(TaskAttemptEventType.TA_CONTAINER_COMPLETED,
      event3.getType());
  TaskAttemptEvent abortedEvent3 = allocator.createContainerFinishedEvent(
      killedByContainerSchedulerStatus, attemptId);
  Assert.assertEquals(TaskAttemptEventType.TA_KILL, abortedEvent3.getType());
}