LocalContainerLauncher extends AbstractService implements
    ContainerLauncher { @Override public void handle(ContainerLauncherEvent event) { try { eventQueue.put(event); } catch (InterruptedException e) { throw new YarnRuntimeException(e); } } LocalContainerLauncher(AppContext context,
                                TaskUmbilicalProtocol umbilical);  LocalContainerLauncher(AppContext context,
                                TaskUmbilicalProtocol umbilical,
                                ClassLoader jobClassLoader); void serviceStart(); void serviceStop(); @Override void handle(ContainerLauncherEvent event); void setEncryptedSpillKey(byte[] encryptedSpillKey);  }
@SuppressWarnings("rawtypes") @Test(timeout=10000) public void testKillJob() throws Exception { JobConf conf = new JobConf(); AppContext context = mock(AppContext.class); final CountDownLatch isDone = new CountDownLatch(1); EventHandler handler = new EventHandler() { @Override public void handle(Event event) { LOG.info("handling event " + event.getClass() + " with type " + event.getType()); if (event instanceof TaskAttemptEvent) { if (event.getType() == TaskAttemptEventType.TA_CONTAINER_CLEANED) { isDone.countDown(); } } } }; when(context.getEventHandler()).thenReturn(handler); LocalContainerLauncher launcher = new LocalContainerLauncher(context, mock(TaskUmbilicalProtocol.class)); launcher.init(conf); launcher.start(); JobId jobId = MRBuilderUtils.newJobId(System.currentTimeMillis(), 1, 1); TaskId taskId = MRBuilderUtils.newTaskId(jobId, 1, TaskType.MAP); TaskAttemptId taId = MRBuilderUtils.newTaskAttemptId(taskId, 0); Job job = mock(Job.class); when(job.getTotalMaps()).thenReturn(1); when(job.getTotalReduces()).thenReturn(0); Map<JobId,Job> jobs = new HashMap<JobId,Job>(); jobs.put(jobId, job); when(context.getAllJobs()).thenReturn(jobs); org.apache.hadoop.mapreduce.v2.app.job.Task ytask = mock(org.apache.hadoop.mapreduce.v2.app.job.Task.class); when(ytask.getType()).thenReturn(TaskType.MAP); when(job.getTask(taskId)).thenReturn(ytask); MapTask mapTask = mock(MapTask.class); when(mapTask.isMapOrReduce()).thenReturn(true); when(mapTask.isMapTask()).thenReturn(true); TaskAttemptID taskID = TypeConverter.fromYarn(taId); when(mapTask.getTaskID()).thenReturn(taskID); when(mapTask.getJobID()).thenReturn(taskID.getJobID()); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { LOG.info("sleeping for 5 minutes..."); Thread.sleep(5*60*1000); return null; } }).when(mapTask).run(isA(JobConf.class), isA(TaskUmbilicalProtocol.class)); ContainerLauncherEvent launchEvent = new ContainerRemoteLaunchEvent(taId, null, createMockContainer(), mapTask); launcher.handle(launchEvent); Thread.sleep(200); ContainerLauncherEvent cleanupEvent = new ContainerLauncherEvent(taId, null, null, null, ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP); launcher.handle(cleanupEvent); isDone.await(); launcher.close(); }