@Override
  public void handle(CommitterEvent event) {
    try {
      eventQueue.put(event);
    } catch (InterruptedException e) {
      throw new YarnRuntimeException(e);
    }
  }
@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();

    // 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());

    //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());

    ceh.stop();
    dispatcher.stop();
  }