ContainersLauncher extends AbstractService implements EventHandler<ContainersLauncherEvent> { @Override public void handle(ContainersLauncherEvent event) { Container container = event.getContainer(); ContainerId containerId = container.getContainerId(); switch (event.getType()) { case LAUNCH_CONTAINER: Application app = context.getApplications().get( containerId.getApplicationAttemptId().getApplicationId()); ContainerLaunch launch = new ContainerLaunch(context, getConfig(), dispatcher, exec, app, event.getContainer(), dirsHandler, containerManager); containerLauncher.submit(launch); running.put(containerId, launch); break; case RELAUNCH_CONTAINER: app = context.getApplications().get( containerId.getApplicationAttemptId().getApplicationId()); ContainerRelaunch relaunch = new ContainerRelaunch(context, getConfig(), dispatcher, exec, app, event.getContainer(), dirsHandler, containerManager); containerLauncher.submit(relaunch); running.put(containerId, relaunch); break; case RECOVER_CONTAINER: app = context.getApplications().get( containerId.getApplicationAttemptId().getApplicationId()); launch = new RecoveredContainerLaunch(context, getConfig(), dispatcher, exec, app, event.getContainer(), dirsHandler, containerManager); containerLauncher.submit(launch); running.put(containerId, launch); break; case RECOVER_PAUSED_CONTAINER: app = context.getApplications().get( containerId.getApplicationAttemptId().getApplicationId()); launch = new RecoverPausedContainerLaunch(context, getConfig(), dispatcher, exec, app, event.getContainer(), dirsHandler, containerManager); containerLauncher.submit(launch); break; case CLEANUP_CONTAINER: case CLEANUP_CONTAINER_FOR_REINIT: ContainerLaunch existingLaunch = running.remove(containerId); if (existingLaunch == null) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerId, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, Shell.WINDOWS ? ContainerExecutor.ExitCode.FORCE_KILLED.getExitCode() : ContainerExecutor.ExitCode.TERMINATED.getExitCode(), "Container terminated before launch.")); return; } ContainerCleanup cleanup = new ContainerCleanup(context, getConfig(), dispatcher, exec, event.getContainer(), existingLaunch); containerLauncher.submit(cleanup); break; case SIGNAL_CONTAINER: SignalContainersLauncherEvent signalEvent = (SignalContainersLauncherEvent) event; ContainerLaunch runningContainer = running.get(containerId); if (runningContainer == null) { LOG.info("Container " + containerId + " not running, nothing to signal."); return; } try { runningContainer.signalContainer(signalEvent.getCommand()); } catch (IOException e) { LOG.warn("Got exception while signaling container " + containerId + " with command " + signalEvent.getCommand()); } break; case PAUSE_CONTAINER: ContainerLaunch launchedContainer = running.get(containerId); if (launchedContainer == null) { return; } try { launchedContainer.pauseContainer(); } catch (Exception e) { LOG.info("Got exception while pausing container: " + StringUtils.stringifyException(e)); } break; case RESUME_CONTAINER: ContainerLaunch launchCont = running.get(containerId); if (launchCont == null) { return; } try { launchCont.resumeContainer(); } catch (Exception e) { LOG.info("Got exception while resuming container: " + StringUtils.stringifyException(e)); } break; } } ContainersLauncher(Context context, Dispatcher dispatcher,
      ContainerExecutor exec, LocalDirsHandlerService dirsHandler,
      ContainerManagerImpl containerManager); @Override void handle(ContainersLauncherEvent event); @VisibleForTesting
  public ExecutorService containerLauncher; @VisibleForTesting
  final Map<ContainerId, ContainerLaunch> running; }
@SuppressWarnings("unchecked") @Test public void testLaunchContainerEvent() throws IllegalArgumentException, IllegalAccessException { Map<ContainerId, ContainerLaunch> dummyMap = (Map<ContainerId, ContainerLaunch>) Whitebox.getInternalState(spy, "running"); when(event.getType()) .thenReturn(ContainersLauncherEventType.LAUNCH_CONTAINER); assertEquals(0, dummyMap.size()); spy.handle(event); assertEquals(1, dummyMap.size()); Mockito.verify(containerLauncher, Mockito.times(1)) .submit(Mockito.any(ContainerLaunch.class)); }
@SuppressWarnings("unchecked") @Test public void testRelaunchContainerEvent() throws IllegalArgumentException, IllegalAccessException { Map<ContainerId, ContainerLaunch> dummyMap = (Map<ContainerId, ContainerLaunch>) Whitebox.getInternalState(spy, "running"); when(event.getType()) .thenReturn(ContainersLauncherEventType.RELAUNCH_CONTAINER); assertEquals(0, dummyMap.size()); spy.handle(event); assertEquals(1, dummyMap.size()); Mockito.verify(containerLauncher, Mockito.times(1)) .submit(Mockito.any(ContainerRelaunch.class)); for (ContainerId cid : dummyMap.keySet()) { Object o = dummyMap.get(cid); assertEquals(true, (o instanceof ContainerRelaunch)); } }
@SuppressWarnings("unchecked") @Test public void testRecoverContainerEvent() throws IllegalArgumentException, IllegalAccessException { Map<ContainerId, ContainerLaunch> dummyMap = (Map<ContainerId, ContainerLaunch>) Whitebox.getInternalState(spy, "running"); when(event.getType()) .thenReturn(ContainersLauncherEventType.RECOVER_CONTAINER); assertEquals(0, dummyMap.size()); spy.handle(event); assertEquals(1, dummyMap.size()); Mockito.verify(containerLauncher, Mockito.times(1)) .submit(Mockito.any(RecoveredContainerLaunch.class)); for (ContainerId cid : dummyMap.keySet()) { Object o = dummyMap.get(cid); assertEquals(true, (o instanceof RecoveredContainerLaunch)); } }
@Test public void testRecoverPausedContainerEvent() throws IllegalArgumentException, IllegalAccessException { when(event.getType()) .thenReturn(ContainersLauncherEventType.RECOVER_PAUSED_CONTAINER); spy.handle(event); Mockito.verify(containerLauncher, Mockito.times(1)) .submit(Mockito.any(RecoverPausedContainerLaunch.class)); }
@Test public void testCleanupContainerEvent() throws IllegalArgumentException, IllegalAccessException, IOException { Map<ContainerId, ContainerLaunch> dummyMap = Collections .synchronizedMap(new HashMap<ContainerId, ContainerLaunch>()); dummyMap.put(containerId, containerLaunch); Whitebox.setInternalState(spy, "running", dummyMap); when(event.getType()) .thenReturn(ContainersLauncherEventType.CLEANUP_CONTAINER); assertEquals(1, dummyMap.size()); spy.handle(event); assertEquals(0, dummyMap.size()); Mockito.verify(containerLauncher, Mockito.times(1)) .submit(Mockito.any(ContainerCleanup.class)); }
@Test public void testCleanupContainerForReINITEvent() throws IllegalArgumentException, IllegalAccessException, IOException { Map<ContainerId, ContainerLaunch> dummyMap = Collections .synchronizedMap(new HashMap<ContainerId, ContainerLaunch>()); dummyMap.put(containerId, containerLaunch); Whitebox.setInternalState(spy, "running", dummyMap); when(event.getType()) .thenReturn(ContainersLauncherEventType.CLEANUP_CONTAINER_FOR_REINIT); assertEquals(1, dummyMap.size()); spy.handle(event); assertEquals(0, dummyMap.size()); Mockito.verify(containerLauncher, Mockito.times(1)) .submit(Mockito.any(ContainerCleanup.class)); }
@Test public void testSignalContainerEvent() throws IllegalArgumentException, IllegalAccessException, IOException { Map<ContainerId, ContainerLaunch> dummyMap = Collections .synchronizedMap(new HashMap<ContainerId, ContainerLaunch>()); dummyMap.put(containerId, containerLaunch); SignalContainersLauncherEvent dummyEvent = mock(SignalContainersLauncherEvent.class); when(dummyEvent.getContainer()).thenReturn(container); when(container.getContainerId()).thenReturn(containerId); when(containerId.getApplicationAttemptId()).thenReturn(appAttemptId); when(containerId.getApplicationAttemptId().getApplicationId()) .thenReturn(appId); Whitebox.setInternalState(spy, "running", dummyMap); when(dummyEvent.getType()) .thenReturn(ContainersLauncherEventType.SIGNAL_CONTAINER); when(dummyEvent.getCommand()) .thenReturn(SignalContainerCommand.GRACEFUL_SHUTDOWN); doNothing().when(containerLaunch) .signalContainer(SignalContainerCommand.GRACEFUL_SHUTDOWN); spy.handle(dummyEvent); assertEquals(1, dummyMap.size()); Mockito.verify(containerLaunch, Mockito.times(1)) .signalContainer(SignalContainerCommand.GRACEFUL_SHUTDOWN); }
@Test public void testPauseContainerEvent() throws IllegalArgumentException, IllegalAccessException, IOException { Map<ContainerId, ContainerLaunch> dummyMap = Collections .synchronizedMap(new HashMap<ContainerId, ContainerLaunch>()); dummyMap.put(containerId, containerLaunch); Whitebox.setInternalState(spy, "running", dummyMap); when(event.getType()) .thenReturn(ContainersLauncherEventType.PAUSE_CONTAINER); doNothing().when(containerLaunch).pauseContainer(); spy.handle(event); assertEquals(1, dummyMap.size()); Mockito.verify(containerLaunch, Mockito.times(1)).pauseContainer(); }
@Test public void testResumeContainerEvent() throws IllegalArgumentException, IllegalAccessException, IOException { Map<ContainerId, ContainerLaunch> dummyMap = Collections .synchronizedMap(new HashMap<ContainerId, ContainerLaunch>()); dummyMap.put(containerId, containerLaunch); Whitebox.setInternalState(spy, "running", dummyMap); when(event.getType()) .thenReturn(ContainersLauncherEventType.RESUME_CONTAINER); doNothing().when(containerLaunch).resumeContainer(); spy.handle(event); assertEquals(1, dummyMap.size()); Mockito.verify(containerLaunch, Mockito.times(1)).resumeContainer(); }