@Private
  public void handleEvent(JobHistoryEvent event) {
    synchronized (lock) {

      // If this is JobSubmitted Event, setup the writer
      if (event.getHistoryEvent().getEventType() == EventType.AM_STARTED) {
        try {
          AMStartedEvent amStartedEvent =
              (AMStartedEvent) event.getHistoryEvent();
          setupEventWriter(event.getJobID(), amStartedEvent);
        } catch (IOException ioe) {
          LOG.error("Error JobHistoryEventHandler in handleEvent: " + event,
              ioe);
          throw new YarnRuntimeException(ioe);
        }
      }

      // For all events
      // (1) Write it out
      // (2) Process it for JobSummary
      // (3) Process it for ATS (if enabled)
      MetaInfo mi = fileMap.get(event.getJobID());
      try {
        HistoryEvent historyEvent = event.getHistoryEvent();
        if (! (historyEvent instanceof NormalizedResourceEvent)) {
          mi.writeEvent(historyEvent);
        }
        processEventForJobSummary(event.getHistoryEvent(), mi.getJobSummary(),
            event.getJobID());
        if (LOG.isDebugEnabled()) {
          LOG.debug("In HistoryEventHandler "
              + event.getHistoryEvent().getEventType());
        }
      } catch (IOException e) {
        LOG.error("Error writing History Event: " + event.getHistoryEvent(),
            e);
        throw new YarnRuntimeException(e);
      }

      if (event.getHistoryEvent().getEventType() == EventType.JOB_SUBMITTED) {
        JobSubmittedEvent jobSubmittedEvent =
            (JobSubmittedEvent) event.getHistoryEvent();
        mi.getJobIndexInfo().setSubmitTime(jobSubmittedEvent.getSubmitTime());
        mi.getJobIndexInfo().setQueueName(jobSubmittedEvent.getJobQueueName());
      }
      //initialize the launchTime in the JobIndexInfo of MetaInfo
      if(event.getHistoryEvent().getEventType() == EventType.JOB_INITED ){
        JobInitedEvent jie = (JobInitedEvent) event.getHistoryEvent();
        mi.getJobIndexInfo().setJobStartTime(jie.getLaunchTime());
      }
      
      if (event.getHistoryEvent().getEventType() == EventType.JOB_QUEUE_CHANGED) {
        JobQueueChangeEvent jQueueEvent =
            (JobQueueChangeEvent) event.getHistoryEvent();
        mi.getJobIndexInfo().setQueueName(jQueueEvent.getJobQueueName());
      }

      // If this is JobFinishedEvent, close the writer and setup the job-index
      if (event.getHistoryEvent().getEventType() == EventType.JOB_FINISHED) {
        try {
          JobFinishedEvent jFinishedEvent =
              (JobFinishedEvent) event.getHistoryEvent();
          mi.getJobIndexInfo().setFinishTime(jFinishedEvent.getFinishTime());
          mi.getJobIndexInfo().setNumMaps(jFinishedEvent.getSucceededMaps());
          mi.getJobIndexInfo().setNumReduces(
              jFinishedEvent.getSucceededReduces());
          mi.getJobIndexInfo().setJobStatus(JobState.SUCCEEDED.toString());
          closeEventWriter(event.getJobID());
          processDoneFiles(event.getJobID());
        } catch (IOException e) {
          throw new YarnRuntimeException(e);
        }
      }
      // In case of JOB_ERROR, only process all the Done files(e.g. job
      // summary, job history file etc.) if it is last AM retry.
      if (event.getHistoryEvent().getEventType() == EventType.JOB_ERROR) {
        try {
          JobUnsuccessfulCompletionEvent jucEvent =
              (JobUnsuccessfulCompletionEvent) event.getHistoryEvent();
          mi.getJobIndexInfo().setFinishTime(jucEvent.getFinishTime());
          mi.getJobIndexInfo().setNumMaps(jucEvent.getSucceededMaps());
          mi.getJobIndexInfo().setNumReduces(jucEvent.getSucceededReduces());
          mi.getJobIndexInfo().setJobStatus(jucEvent.getStatus());
          closeEventWriter(event.getJobID());
          if(context.isLastAMRetry())
            processDoneFiles(event.getJobID());
        } catch (IOException e) {
          throw new YarnRuntimeException(e);
        }
      }

      if (event.getHistoryEvent().getEventType() == EventType.JOB_FAILED
          || event.getHistoryEvent().getEventType() == EventType.JOB_KILLED) {
        try {
          JobUnsuccessfulCompletionEvent jucEvent = 
              (JobUnsuccessfulCompletionEvent) event
              .getHistoryEvent();
          mi.getJobIndexInfo().setFinishTime(jucEvent.getFinishTime());
          mi.getJobIndexInfo().setNumMaps(jucEvent.getSucceededMaps());
          mi.getJobIndexInfo().setNumReduces(jucEvent.getSucceededReduces());
          mi.getJobIndexInfo().setJobStatus(jucEvent.getStatus());
          closeEventWriter(event.getJobID());
          processDoneFiles(event.getJobID());
        } catch (IOException e) {
          throw new YarnRuntimeException(e);
        }
      }
    }
  }
@Test (timeout=50000)
  public void testBatchedFlushJobEndMultiplier() 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); //2 seconds.
    conf.setInt(MRJobConfig.MR_AM_HISTORY_JOB_COMPLETE_UNFLUSHED_MULTIPLIER, 3);
    conf.setInt(MRJobConfig.MR_AM_HISTORY_MAX_UNFLUSHED_COMPLETE_EVENTS, 10);
    conf.setInt(
        MRJobConfig.MR_AM_HISTORY_USE_BATCHED_FLUSH_QUEUE_SIZE_THRESHOLD, 0);

    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, -1)));
      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, 0)));
      }
      queueEvent(jheh, new JobHistoryEvent(t.jobId, new JobFinishedEvent(
          TypeConverter.fromYarn(t.jobId), 0, 10, 10, 0, 0, 0, 0, null, null,
          new Counters())));

      handleNextNEvents(jheh, 29);
      verify(mockWriter, times(0)).flush();

      handleNextNEvents(jheh, 72);
      verify(mockWriter, times(4)).flush(); //3 * 30 + 1 for JobFinished
    } finally {
      jheh.stop();
      verify(mockWriter).close();
    }
  }