@Test (timeout = 5000) public void testContainerPreempted() throws IOException { Configuration conf = new Configuration(false); schedulerHandler.init(conf); schedulerHandler.start(); String diagnostics = "Container preempted by RM."; TaskAttemptImpl mockTask = mock(TaskAttemptImpl.class); ContainerStatus mockStatus = mock(ContainerStatus.class); ContainerId mockCId = mock(ContainerId.class); AMContainer mockAMContainer = mock(AMContainer.class); when(mockAMContainerMap.get(mockCId)).thenReturn(mockAMContainer); when(mockAMContainer.getContainerId()).thenReturn(mockCId); when(mockStatus.getContainerId()).thenReturn(mockCId); when(mockStatus.getDiagnostics()).thenReturn(diagnostics); when(mockStatus.getExitStatus()).thenReturn(ContainerExitStatus.PREEMPTED); schedulerHandler.containerCompleted(mockTask, mockStatus); Assert.assertEquals(1, mockEventHandler.events.size()); Event event = mockEventHandler.events.get(0); Assert.assertEquals(AMContainerEventType.C_COMPLETED, event.getType()); AMContainerEventCompleted completedEvent = (AMContainerEventCompleted) event; Assert.assertEquals(mockCId, completedEvent.getContainerId()); Assert.assertEquals("Container preempted externally. Container preempted by RM.", completedEvent.getDiagnostics()); Assert.assertTrue(completedEvent.isPreempted()); Assert.assertFalse(completedEvent.isDiskFailed()); schedulerHandler.stop(); schedulerHandler.close(); }
@Override public synchronized void containerCompleted(Object task, ContainerStatus containerStatus) { AMContainer amContainer = appContext.getAllContainers().get(containerStatus.getContainerId()); if (amContainer != null) { String message = null; int exitStatus = containerStatus.getExitStatus(); if (exitStatus == ContainerExitStatus.PREEMPTED) { message = "Container preempted externally. "; } else if (exitStatus == ContainerExitStatus.DISKS_FAILED) { message = "Container disk failed. "; } else { message = "Container failed. "; } if (containerStatus.getDiagnostics() != null) { message += containerStatus.getDiagnostics(); } sendEvent(new AMContainerEventCompleted(amContainer.getContainerId(), exitStatus, message)); } }
TaskSchedulerEventHandler extends AbstractService implements TaskSchedulerAppCallback,
                                               EventHandler<AMSchedulerEvent> { @Override public synchronized void containerCompleted(Object task, ContainerStatus containerStatus) { AMContainer amContainer = appContext.getAllContainers().get(containerStatus.getContainerId()); if (amContainer != null) { String message = null; int exitStatus = containerStatus.getExitStatus(); if (exitStatus == ContainerExitStatus.PREEMPTED) { message = "Container preempted externally. "; } else if (exitStatus == ContainerExitStatus.DISKS_FAILED) { message = "Container disk failed. "; } else { message = "Container failed. "; } if (containerStatus.getDiagnostics() != null) { message += containerStatus.getDiagnostics(); } sendEvent(new AMContainerEventCompleted(amContainer.getContainerId(), exitStatus, message)); } } }
TaskSchedulerEventHandler extends AbstractService implements TaskSchedulerAppCallback,
                                               EventHandler<AMSchedulerEvent> { @Override public synchronized void containerCompleted(Object task, ContainerStatus containerStatus) { AMContainer amContainer = appContext.getAllContainers().get(containerStatus.getContainerId()); if (amContainer != null) { String message = null; int exitStatus = containerStatus.getExitStatus(); if (exitStatus == ContainerExitStatus.PREEMPTED) { message = "Container preempted externally. "; } else if (exitStatus == ContainerExitStatus.DISKS_FAILED) { message = "Container disk failed. "; } else { message = "Container failed. "; } if (containerStatus.getDiagnostics() != null) { message += containerStatus.getDiagnostics(); } sendEvent(new AMContainerEventCompleted(amContainer.getContainerId(), exitStatus, message)); } } @SuppressWarnings("rawtypes")  TaskSchedulerEventHandler(AppContext appContext,
      DAGClientServer clientService, EventHandler eventHandler, ContainerSignatureMatcher containerSignatureMatcher); }
TaskSchedulerEventHandler extends AbstractService implements TaskSchedulerAppCallback,
                                               EventHandler<AMSchedulerEvent> { @Override public synchronized void containerCompleted(Object task, ContainerStatus containerStatus) { AMContainer amContainer = appContext.getAllContainers().get(containerStatus.getContainerId()); if (amContainer != null) { String message = null; int exitStatus = containerStatus.getExitStatus(); if (exitStatus == ContainerExitStatus.PREEMPTED) { message = "Container preempted externally. "; } else if (exitStatus == ContainerExitStatus.DISKS_FAILED) { message = "Container disk failed. "; } else { message = "Container failed. "; } if (containerStatus.getDiagnostics() != null) { message += containerStatus.getDiagnostics(); } sendEvent(new AMContainerEventCompleted(amContainer.getContainerId(), exitStatus, message)); } } @SuppressWarnings("rawtypes")  TaskSchedulerEventHandler(AppContext appContext,
      DAGClientServer clientService, EventHandler eventHandler, ContainerSignatureMatcher containerSignatureMatcher); Map<ApplicationAccessType, String> getApplicationAcls(); void setSignalled(boolean isSignalled); int getNumClusterNodes(); Resource getAvailableResources(); Resource getTotalResources(); synchronized void handleEvent(AMSchedulerEvent sEvent); @Override void handle(AMSchedulerEvent event); @Override synchronized void serviceStart(); @Override void serviceStop(); @Override synchronized void taskAllocated(Object task,
                                           Object appCookie,
                                           Container container); @Override synchronized void containerCompleted(Object task, ContainerStatus containerStatus); @Override synchronized void containerBeingReleased(ContainerId containerId); @SuppressWarnings("unchecked") @Override synchronized void nodesUpdated(List<NodeReport> updatedNodes); @Override synchronized void appShutdownRequested(); @Override synchronized void setApplicationRegistrationData(
      Resource maxContainerCapability,
      Map<ApplicationAccessType, String> appAcls, 
      ByteBuffer clientAMSecretKey); @Override AppFinalStatus getFinalAppStatus(); @Override float getProgress(); @Override void onError(Throwable t); void dagCompleted(); @Override void preemptContainer(ContainerId containerId); void setShouldUnregisterFlag(); }
TaskSchedulerEventHandler extends AbstractService implements TaskSchedulerAppCallback,
                                               EventHandler<AMSchedulerEvent> { @Override public synchronized void containerCompleted(Object task, ContainerStatus containerStatus) { AMContainer amContainer = appContext.getAllContainers().get(containerStatus.getContainerId()); if (amContainer != null) { String message = null; int exitStatus = containerStatus.getExitStatus(); if (exitStatus == ContainerExitStatus.PREEMPTED) { message = "Container preempted externally. "; } else if (exitStatus == ContainerExitStatus.DISKS_FAILED) { message = "Container disk failed. "; } else { message = "Container failed. "; } if (containerStatus.getDiagnostics() != null) { message += containerStatus.getDiagnostics(); } sendEvent(new AMContainerEventCompleted(amContainer.getContainerId(), exitStatus, message)); } } @SuppressWarnings("rawtypes")  TaskSchedulerEventHandler(AppContext appContext,
      DAGClientServer clientService, EventHandler eventHandler, ContainerSignatureMatcher containerSignatureMatcher); Map<ApplicationAccessType, String> getApplicationAcls(); void setSignalled(boolean isSignalled); int getNumClusterNodes(); Resource getAvailableResources(); Resource getTotalResources(); synchronized void handleEvent(AMSchedulerEvent sEvent); @Override void handle(AMSchedulerEvent event); @Override synchronized void serviceStart(); @Override void serviceStop(); @Override synchronized void taskAllocated(Object task,
                                           Object appCookie,
                                           Container container); @Override synchronized void containerCompleted(Object task, ContainerStatus containerStatus); @Override synchronized void containerBeingReleased(ContainerId containerId); @SuppressWarnings("unchecked") @Override synchronized void nodesUpdated(List<NodeReport> updatedNodes); @Override synchronized void appShutdownRequested(); @Override synchronized void setApplicationRegistrationData(
      Resource maxContainerCapability,
      Map<ApplicationAccessType, String> appAcls, 
      ByteBuffer clientAMSecretKey); @Override AppFinalStatus getFinalAppStatus(); @Override float getProgress(); @Override void onError(Throwable t); void dagCompleted(); @Override void preemptContainer(ContainerId containerId); void setShouldUnregisterFlag();  }