@Test public void testBasic() throws Exception { AppContext mockContext = mock(AppContext.class); OutputCommitter mockCommitter = mock(OutputCommitter.class); Clock mockClock = mock(Clock.class); CommitterEventHandler handler = new CommitterEventHandler(mockContext, mockCommitter, new TestingRMHeartbeatHandler()); YarnConfiguration conf = new YarnConfiguration(); conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir); JobContext mockJobContext = mock(JobContext.class); ApplicationAttemptId attemptid = ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0"); JobId jobId = TypeConverter.toYarn( TypeConverter.fromYarn(attemptid.getApplicationId())); WaitForItHandler waitForItHandler = new WaitForItHandler(); when(mockContext.getApplicationID()).thenReturn(attemptid.getApplicationId()); when(mockContext.getApplicationAttemptId()).thenReturn(attemptid); when(mockContext.getEventHandler()).thenReturn(waitForItHandler); when(mockContext.getClock()).thenReturn(mockClock); handler.init(conf); handler.start(); try { handler.handle(new CommitterJobCommitEvent(jobId, mockJobContext)); String user = UserGroupInformation.getCurrentUser().getShortUserName(); Path startCommitFile = MRApps.getStartJobCommitFile(conf, user, jobId); Path endCommitSuccessFile = MRApps.getEndJobCommitSuccessFile(conf, user, jobId); Path endCommitFailureFile = MRApps.getEndJobCommitFailureFile(conf, user, jobId); Event e = waitForItHandler.getAndClearEvent(); assertNotNull(e); assertTrue(e instanceof JobCommitCompletedEvent); FileSystem fs = FileSystem.get(conf); assertTrue(startCommitFile.toString(), fs.exists(startCommitFile)); assertTrue(endCommitSuccessFile.toString(), fs.exists(endCommitSuccessFile)); assertFalse(endCommitFailureFile.toString(), fs.exists(endCommitFailureFile)); verify(mockCommitter).commitJob(any(JobContext.class)); } finally { handler.stop(); } }
@Override public void handle(CommitterEvent event) { try { eventQueue.put(event); } catch (InterruptedException e) { throw new YarnRuntimeException(e); } }
CommitterEventHandler extends AbstractService implements EventHandler<CommitterEvent> { @Override public void handle(CommitterEvent event) { try { eventQueue.put(event); } catch (InterruptedException e) { throw new YarnRuntimeException(e); } } }
CommitterEventHandler extends AbstractService implements EventHandler<CommitterEvent> { @Override public void handle(CommitterEvent event) { try { eventQueue.put(event); } catch (InterruptedException e) { throw new YarnRuntimeException(e); } } CommitterEventHandler(AppContext context, OutputCommitter committer,
      RMHeartbeatHandler rmHeartbeatHandler);  CommitterEventHandler(AppContext context, OutputCommitter committer,
      RMHeartbeatHandler rmHeartbeatHandler, ClassLoader jobClassLoader); }
CommitterEventHandler extends AbstractService implements EventHandler<CommitterEvent> { @Override public void handle(CommitterEvent event) { try { eventQueue.put(event); } catch (InterruptedException e) { throw new YarnRuntimeException(e); } } CommitterEventHandler(AppContext context, OutputCommitter committer,
      RMHeartbeatHandler rmHeartbeatHandler);  CommitterEventHandler(AppContext context, OutputCommitter committer,
      RMHeartbeatHandler rmHeartbeatHandler, ClassLoader jobClassLoader); @Override void handle(CommitterEvent event); }
CommitterEventHandler extends AbstractService implements EventHandler<CommitterEvent> { @Override public void handle(CommitterEvent event) { try { eventQueue.put(event); } catch (InterruptedException e) { throw new YarnRuntimeException(e); } } CommitterEventHandler(AppContext context, OutputCommitter committer,
      RMHeartbeatHandler rmHeartbeatHandler);  CommitterEventHandler(AppContext context, OutputCommitter committer,
      RMHeartbeatHandler rmHeartbeatHandler, ClassLoader jobClassLoader); @Override void handle(CommitterEvent event);  }