@VisibleForTesting
  public TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont,
      TaskAttemptId attemptId) {
    TaskAttemptEvent event;
    switch (cont.getExitStatus()) {
    case ContainerExitStatus.ABORTED:
    case ContainerExitStatus.PREEMPTED:
    case ContainerExitStatus.KILLED_BY_CONTAINER_SCHEDULER:
      // killed by YARN
      event = new TaskAttemptEvent(attemptId, TaskAttemptEventType.TA_KILL);
      break;
    default:
      event = new TaskAttemptEvent(attemptId,
          TaskAttemptEventType.TA_CONTAINER_COMPLETED);
    }
    return event;
  }
@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());
  }