@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 = new SystemClock(); AppContext appContext = mock(AppContext.class); ApplicationAttemptId attemptid = ConverterUtils.toApplicationAttemptId("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(JobContext.class)); 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(JobContext.class)); ceh.stop(); dispatcher.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);  }