@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();
  // verify trying to commit when RM heartbeats are stale does not commit
  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);
  // set a fresh heartbeat and verify commit completes
  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));
  //Clean up so we can try to commit again (Don't do this at home)
  cleanup();
  
  // try to commit again and verify it goes through since the heartbeat
  // is still fresh
  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();
}