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);  }
@Test public void testCommitWindow() throws Exception { Configuration conf = new Configuration(); conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir); AsyncDispatcher dispatcher = new AsyncDispatcher(); dispatcher.init(conf); dispatcher.start(); TestingJobEventHandler jeh = new TestingJobEventHandler(); dispatcher.register(JobEventType.class, jeh); SystemClock clock = SystemClock.getInstance(); AppContext appContext = mock(AppContext.class); ApplicationAttemptId attemptid = ApplicationAttemptId.fromString( "appattempt_1234567890000_0001_0"); when(appContext.getApplicationID()).thenReturn(attemptid.getApplicationId()); when(appContext.getApplicationAttemptId()).thenReturn(attemptid); when(appContext.getEventHandler()).thenReturn( dispatcher.getEventHandler()); when(appContext.getClock()).thenReturn(clock); OutputCommitter committer = mock(OutputCommitter.class); TestingRMHeartbeatHandler rmhh = new TestingRMHeartbeatHandler(); CommitterEventHandler ceh = new CommitterEventHandler(appContext, committer, rmhh); ceh.init(conf); ceh.start(); ceh.handle(new CommitterJobCommitEvent(null, null)); long timeToWaitMs = 5000; while (rmhh.getNumCallbacks() != 1 && timeToWaitMs > 0) { Thread.sleep(10); timeToWaitMs -= 10; } Assert.assertEquals("committer did not register a heartbeat callback", 1, rmhh.getNumCallbacks()); verify(committer, never()).commitJob(any(JobContext.class)); Assert.assertEquals("committer should not have committed", 0, jeh.numCommitCompletedEvents); rmhh.setLastHeartbeatTime(clock.getTime()); timeToWaitMs = 5000; while (jeh.numCommitCompletedEvents != 1 && timeToWaitMs > 0) { Thread.sleep(10); timeToWaitMs -= 10; } Assert.assertEquals("committer did not complete commit after RM hearbeat", 1, jeh.numCommitCompletedEvents); verify(committer, times(1)).commitJob(any()); cleanup(); ceh.handle(new CommitterJobCommitEvent(null, null)); timeToWaitMs = 5000; while (jeh.numCommitCompletedEvents != 2 && timeToWaitMs > 0) { Thread.sleep(10); timeToWaitMs -= 10; } Assert.assertEquals("committer did not commit", 2, jeh.numCommitCompletedEvents); verify(committer, times(2)).commitJob(any()); ceh.stop(); dispatcher.stop(); }
@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 = ApplicationAttemptId.fromString( "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(); } }
@Test public void testFailure() 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 = ApplicationAttemptId.fromString("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); doThrow(new YarnRuntimeException("Intentional Failure")).when(mockCommitter) .commitJob(any(JobContext.class)); 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 JobCommitFailedEvent); FileSystem fs = FileSystem.get(conf); assertTrue(fs.exists(startCommitFile)); assertFalse(fs.exists(endCommitSuccessFile)); assertTrue(fs.exists(endCommitFailureFile)); verify(mockCommitter).commitJob(any(JobContext.class)); } finally { handler.stop(); } }