@Test public void testCompletedContainerEvent() { RMContainerAllocator allocator = new RMContainerAllocator( mock(ClientService.class), mock(AppContext.class)); 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); 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()); 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()); }
@VisibleForTesting public TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont, TaskAttemptId attemptID) { if (cont.getExitStatus() == ContainerExitStatus.ABORTED || cont.getExitStatus() == ContainerExitStatus.PREEMPTED) { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_KILL); } else { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_CONTAINER_COMPLETED); } }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { @VisibleForTesting public TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont, TaskAttemptId attemptID) { if (cont.getExitStatus() == ContainerExitStatus.ABORTED || cont.getExitStatus() == ContainerExitStatus.PREEMPTED) { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_KILL); } else { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_CONTAINER_COMPLETED); } } }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { @VisibleForTesting public TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont, TaskAttemptId attemptID) { if (cont.getExitStatus() == ContainerExitStatus.ABORTED || cont.getExitStatus() == ContainerExitStatus.PREEMPTED) { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_KILL); } else { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_CONTAINER_COMPLETED); } } RMContainerAllocator(ClientService clientService, AppContext context); }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { @VisibleForTesting public TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont, TaskAttemptId attemptID) { if (cont.getExitStatus() == ContainerExitStatus.ABORTED || cont.getExitStatus() == ContainerExitStatus.PREEMPTED) { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_KILL); } else { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_CONTAINER_COMPLETED); } } RMContainerAllocator(ClientService clientService, AppContext context); boolean getIsReduceStarted(); void setIsReduceStarted(boolean reduceStarted); @Override void handle(ContainerAllocatorEvent event); @Private void scheduleReduces(
      int totalMaps, int completedMaps,
      int scheduledMaps, int scheduledReduces,
      int assignedMaps, int assignedReduces,
      Resource mapResourceReqt, Resource reduceResourceReqt,
      int numPendingReduces,
      float maxReduceRampupLimit, float reduceSlowStart); @Private void scheduleAllReduces(); @Private void rampUpReduces(int rampUp); @Private void rampDownReduces(int rampDown); @VisibleForTesting TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont,
      TaskAttemptId attemptID); @Private Resource getResourceLimit(); @VisibleForTesting int getNumOfPendingReduces(); }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { @VisibleForTesting public TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont, TaskAttemptId attemptID) { if (cont.getExitStatus() == ContainerExitStatus.ABORTED || cont.getExitStatus() == ContainerExitStatus.PREEMPTED) { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_KILL); } else { return new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_CONTAINER_COMPLETED); } } RMContainerAllocator(ClientService clientService, AppContext context); boolean getIsReduceStarted(); void setIsReduceStarted(boolean reduceStarted); @Override void handle(ContainerAllocatorEvent event); @Private void scheduleReduces(
      int totalMaps, int completedMaps,
      int scheduledMaps, int scheduledReduces,
      int assignedMaps, int assignedReduces,
      Resource mapResourceReqt, Resource reduceResourceReqt,
      int numPendingReduces,
      float maxReduceRampupLimit, float reduceSlowStart); @Private void scheduleAllReduces(); @Private void rampUpReduces(int rampUp); @Private void rampDownReduces(int rampDown); @VisibleForTesting TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont,
      TaskAttemptId attemptID); @Private Resource getResourceLimit(); @VisibleForTesting int getNumOfPendingReduces(); static final float DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART; @VisibleForTesting
  static final String RAMPDOWN_DIAGNOSTIC; }