@SuppressWarnings("unchecked") @Override public void transition(TaskAttemptImpl taskAttempt, TaskAttemptEvent event) { // unregister it to TaskAttemptListener so that it stops listening // for it taskAttempt.taskAttemptListener.unregister( taskAttempt.attemptId, taskAttempt.jvmID); if (event instanceof TaskAttemptKillEvent) { taskAttempt.addDiagnosticInfo( ((TaskAttemptKillEvent) event).getMessage()); } taskAttempt.reportedStatus.progress = 1.0f; taskAttempt.updateProgressSplits(); //send the cleanup event to containerLauncher taskAttempt.eventHandler.handle(new ContainerLauncherEvent( taskAttempt.attemptId, taskAttempt.container.getId(), StringInterner .weakIntern(taskAttempt.container.getNodeId().toString()), taskAttempt.container.getContainerToken(), ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP)); }
@Override protected ContainerLauncher createContainerLauncher(AppContext context) { MockContainerLauncher launcher = new MockContainerLauncher() { @Override public void handle(ContainerLauncherEvent event) { TaskAttemptId taskAttemptID = event.getTaskAttemptID(); // Pass everything except the 2nd attempt of the first task. if (taskAttemptID.getId() != 1 || taskAttemptID.getTaskId().getId() != 0) { super.handle(event); } } }; launcher.shufflePort = 5467; return launcher; }
@Override protected ContainerLauncher createContainerLauncher(AppContext context) { return new MockContainerLauncher() { @Override public void handle(ContainerLauncherEvent event) { if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) { ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event; ContainerLaunchContext launchContext = launchEvent.getContainerLaunchContext(); String cmdString = launchContext.getCommands().toString(); LOG.info("launchContext " + cmdString); myCommandLine = cmdString; cmdEnvironment = launchContext.getEnvironment(); } super.handle(event); } }; }
@SuppressWarnings("unchecked") @Override public void transition(TaskAttemptImpl taskAttempt, TaskAttemptEvent event) { taskAttempt.setFinishTime(); taskAttempt.taskAttemptListener.unregister( taskAttempt.attemptId, taskAttempt.jvmID); taskAttempt.eventHandler.handle(new ContainerLauncherEvent( taskAttempt.attemptId, taskAttempt.getAssignedContainerID(), taskAttempt.getAssignedContainerMgrAddress(), taskAttempt.container.getContainerToken(), ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP)); taskAttempt.eventHandler.handle(new TaskTAttemptEvent( taskAttempt.attemptId, TaskEventType.T_ATTEMPT_KILLED)); }
@Override protected ContainerLauncher createContainerLauncher(AppContext context) { return new MockContainerLauncher() { @Override public void handle(ContainerLauncherEvent event) { if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) { ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event; ContainerLaunchContext launchContext = launchEvent.getContainerLaunchContext(); String cmdString = launchContext.getCommands().toString(); LOG.info("launchContext " + cmdString); launchCmdList.add(cmdString); cmdEnvironment = launchContext.getEnvironment(); } super.handle(event); } }; }
@SuppressWarnings("unchecked") @Override public void transition(TaskAttemptImpl taskAttempt, TaskAttemptEvent event) { // unregister it to TaskAttemptListener so that it stops listening // for it taskAttempt.taskAttemptListener.unregister( taskAttempt.attemptId, taskAttempt.jvmID); taskAttempt.reportedStatus.progress = 1.0f; taskAttempt.updateProgressSplits(); //send the cleanup event to containerLauncher taskAttempt.eventHandler.handle(new ContainerLauncherEvent( taskAttempt.attemptId, taskAttempt.container.getId(), StringInterner .weakIntern(taskAttempt.container.getNodeId().toString()), taskAttempt.container.getContainerToken(), ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP)); }
@Override protected ContainerLauncher createContainerLauncher(AppContext context) { return new MockContainerLauncher() { @Override public void handle(ContainerLauncherEvent event) { if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) { ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event; ContainerLaunchContext launchContext = launchEvent.getContainerLaunchContext(); String cmdString = launchContext.getCommands().toString(); LOG.info("launchContext " + cmdString); myCommandLine = cmdString; } super.handle(event); } }; }
@SuppressWarnings("unchecked") private static void sendContainerCleanup(TaskAttemptImpl taskAttempt, TaskAttemptEvent event) { if (event instanceof TaskAttemptKillEvent) { taskAttempt.addDiagnosticInfo( ((TaskAttemptKillEvent) event).getMessage()); } //send the cleanup event to containerLauncher taskAttempt.eventHandler.handle(new ContainerLauncherEvent( taskAttempt.attemptId, taskAttempt.container.getId(), StringInterner .weakIntern(taskAttempt.container.getNodeId().toString()), taskAttempt.container.getContainerToken(), ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP, event.getType() == TaskAttemptEventType.TA_TIMED_OUT)); }
@Test public void testContainerPassThrough() throws Exception { MRApp app = new MRApp(0, 1, true, this.getClass().getName(), true) { @Override protected ContainerLauncher createContainerLauncher(AppContext context) { return new MockContainerLauncher() { @Override public void handle(ContainerLauncherEvent event) { if (event instanceof ContainerRemoteLaunchEvent) { containerObtainedByContainerLauncher = ((ContainerRemoteLaunchEvent) event).getAllocatedContainer(); } super.handle(event); } }; }; }; Job job = app.submit(new Configuration()); app.waitForState(job, JobState.SUCCEEDED); app.verifyCompleted(); Collection<Task> tasks = job.getTasks().values(); Collection<TaskAttempt> taskAttempts = tasks.iterator().next().getAttempts().values(); TaskAttemptImpl taskAttempt = (TaskAttemptImpl) taskAttempts.iterator().next(); // Container from RM should pass through to the launcher. Container object // should be the same. Assert.assertTrue(taskAttempt.container == containerObtainedByContainerLauncher); }
@Override protected ContainerLauncher createContainerLauncher(AppContext context) { return new ContainerLauncherImpl(context) { @Override public void handle(ContainerLauncherEvent event) { switch (event.getType()) { case CONTAINER_REMOTE_LAUNCH: super.handle(event); // Unused event and container. break; case CONTAINER_REMOTE_CLEANUP: getContext().getEventHandler().handle( new TaskAttemptEvent(event.getTaskAttemptID(), TaskAttemptEventType.TA_CONTAINER_CLEANED)); break; } } @Override public ContainerManagementProtocolProxyData getCMProxy( String containerMgrBindAddr, ContainerId containerId) throws IOException { try { synchronized (this) { wait(); // Just hang the thread simulating a very slow NM. } } catch (InterruptedException e) { e.printStackTrace(); } return null; } }; }
@SuppressWarnings("unchecked") private static void sendContainerCompleted(TaskAttemptImpl taskAttempt) { taskAttempt.eventHandler.handle(new ContainerLauncherEvent( taskAttempt.attemptId, taskAttempt.container.getId(), StringInterner .weakIntern(taskAttempt.container.getNodeId().toString()), taskAttempt.container.getContainerToken(), ContainerLauncher.EventType.CONTAINER_COMPLETED)); }
@SuppressWarnings("unchecked") private static void sendContainerCleanup(TaskAttemptImpl taskAttempt, TaskAttemptEvent event) { if (event instanceof TaskAttemptKillEvent) { taskAttempt.addDiagnosticInfo( ((TaskAttemptKillEvent) event).getMessage()); } //send the cleanup event to containerLauncher taskAttempt.eventHandler.handle(new ContainerLauncherEvent( taskAttempt.attemptId, taskAttempt.container.getId(), StringInterner .weakIntern(taskAttempt.container.getNodeId().toString()), taskAttempt.container.getContainerToken(), ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP)); }
@Override protected ContainerLauncher createContainerLauncher(AppContext context) { return new ContainerLauncherImpl(context) { @Override public void handle(ContainerLauncherEvent event) { switch (event.getType()) { case CONTAINER_REMOTE_LAUNCH: super.handle(event); // Unused event and container. break; case CONTAINER_REMOTE_CLEANUP: getContext().getEventHandler().handle( new TaskAttemptEvent(event.getTaskAttemptID(), TaskAttemptEventType.TA_CONTAINER_CLEANED)); break; case CONTAINER_COMPLETED: super.handle(event); } } @Override public ContainerManagementProtocolProxyData getCMProxy( String containerMgrBindAddr, ContainerId containerId) throws IOException { try { synchronized (this) { wait(); // Just hang the thread simulating a very slow NM. } } catch (InterruptedException e) { e.printStackTrace(); } return null; } }; }