@Test (timeout=50000)
public void testMaxUnflushedCompletionEvents() throws Exception {

  TestParams t = new TestParams();
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, t.workDir);
  conf.setLong(MRJobConfig.MR_AM_HISTORY_COMPLETE_EVENT_FLUSH_TIMEOUT_MS,
      60 * 1000l);
  conf.setInt(MRJobConfig.MR_AM_HISTORY_JOB_COMPLETE_UNFLUSHED_MULTIPLIER, 10);
  conf.setInt(MRJobConfig.MR_AM_HISTORY_MAX_UNFLUSHED_COMPLETE_EVENTS, 10);
  conf.setInt(
      MRJobConfig.MR_AM_HISTORY_USE_BATCHED_FLUSH_QUEUE_SIZE_THRESHOLD, 5);
  JHEvenHandlerForTest realJheh =
      new JHEvenHandlerForTest(t.mockAppContext, 0);
  JHEvenHandlerForTest jheh = spy(realJheh);
  jheh.init(conf);
  EventWriter mockWriter = null;
  try {
    jheh.start();
    handleEvent(jheh, new JobHistoryEvent(t.jobId, new AMStartedEvent(
        t.appAttemptId, 200, t.containerId, "nmhost", 3000, 4000)));
    mockWriter = jheh.getEventWriter();
    verify(mockWriter).write(any(HistoryEvent.class));
    for (int i = 0 ; i < 100 ; i++) {
      queueEvent(jheh, new JobHistoryEvent(t.jobId, new TaskFinishedEvent(
          t.taskID, t.taskAttemptID, 0, TaskType.MAP, "", null)));
    }
    handleNextNEvents(jheh, 9);
    verify(mockWriter, times(0)).flush();
    handleNextNEvents(jheh, 1);
    verify(mockWriter).flush();
    handleNextNEvents(jheh, 50);
    verify(mockWriter, times(6)).flush();
  } finally {
    jheh.stop();
    verify(mockWriter).close();
  }
}