@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 testTimelineEventHandling() throws Exception {
    TestParams t = new TestParams(RunningAppContext.class, false);
    Configuration conf = new YarnConfiguration();
    conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
    long currentTime = System.currentTimeMillis();
    try (MiniYARNCluster yarnCluster = new MiniYARNCluster(
        TestJobHistoryEventHandler.class.getSimpleName(), 1, 1, 1, 1)) {
      yarnCluster.init(conf);
      yarnCluster.start();
      Configuration confJHEH = new YarnConfiguration(conf);
      confJHEH.setBoolean(MRJobConfig.MAPREDUCE_JOB_EMIT_TIMELINE_DATA, true);
      confJHEH.set(YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS,
          MiniYARNCluster.getHostname() + ":" +
          yarnCluster.getApplicationHistoryServer().getPort());
      JHEvenHandlerForTest jheh = new JHEvenHandlerForTest(t.mockAppContext, 0);
      jheh.init(confJHEH);
      jheh.start();
      TimelineStore ts = yarnCluster.getApplicationHistoryServer()
              .getTimelineStore();

      handleEvent(jheh, new JobHistoryEvent(t.jobId, new AMStartedEvent(
              t.appAttemptId, 200, t.containerId, "nmhost", 3000, 4000, -1),
              currentTime - 10));
      jheh.getDispatcher().await();
      TimelineEntities entities = ts.getEntities("MAPREDUCE_JOB", null, null,
              null, null, null, null, null, null, null);
      Assert.assertEquals(1, entities.getEntities().size());
      TimelineEntity tEntity = entities.getEntities().get(0);
      Assert.assertEquals(t.jobId.toString(), tEntity.getEntityId());
      Assert.assertEquals(1, tEntity.getEvents().size());
      Assert.assertEquals(EventType.AM_STARTED.toString(),
              tEntity.getEvents().get(0).getEventType());
      Assert.assertEquals(currentTime - 10,
              tEntity.getEvents().get(0).getTimestamp());

      handleEvent(jheh, new JobHistoryEvent(t.jobId,
              new JobSubmittedEvent(TypeConverter.fromYarn(t.jobId), "name",
              "user", 200, "/foo/job.xml",
              new HashMap<JobACL, AccessControlList>(), "default"),
              currentTime + 10));
      jheh.getDispatcher().await();
      entities = ts.getEntities("MAPREDUCE_JOB", null, null, null,
              null, null, null, null, null, null);
      Assert.assertEquals(1, entities.getEntities().size());
      tEntity = entities.getEntities().get(0);
      Assert.assertEquals(t.jobId.toString(), tEntity.getEntityId());
      Assert.assertEquals(2, tEntity.getEvents().size());
      Assert.assertEquals(EventType.JOB_SUBMITTED.toString(),
              tEntity.getEvents().get(0).getEventType());
      Assert.assertEquals(EventType.AM_STARTED.toString(),
              tEntity.getEvents().get(1).getEventType());
      Assert.assertEquals(currentTime + 10,
              tEntity.getEvents().get(0).getTimestamp());
      Assert.assertEquals(currentTime - 10,
              tEntity.getEvents().get(1).getTimestamp());

      handleEvent(jheh, new JobHistoryEvent(t.jobId,
              new JobQueueChangeEvent(TypeConverter.fromYarn(t.jobId), "q2"),
              currentTime - 20));
      jheh.getDispatcher().await();
      entities = ts.getEntities("MAPREDUCE_JOB", null, null, null,
              null, null, null, null, null, null);
      Assert.assertEquals(1, entities.getEntities().size());
      tEntity = entities.getEntities().get(0);
      Assert.assertEquals(t.jobId.toString(), tEntity.getEntityId());
      Assert.assertEquals(3, tEntity.getEvents().size());
      Assert.assertEquals(EventType.JOB_SUBMITTED.toString(),
              tEntity.getEvents().get(0).getEventType());
      Assert.assertEquals(EventType.AM_STARTED.toString(),
              tEntity.getEvents().get(1).getEventType());
      Assert.assertEquals(EventType.JOB_QUEUE_CHANGED.toString(),
              tEntity.getEvents().get(2).getEventType());
      Assert.assertEquals(currentTime + 10,
              tEntity.getEvents().get(0).getTimestamp());
      Assert.assertEquals(currentTime - 10,
              tEntity.getEvents().get(1).getTimestamp());
      Assert.assertEquals(currentTime - 20,
              tEntity.getEvents().get(2).getTimestamp());

      handleEvent(jheh, new JobHistoryEvent(t.jobId,
              new JobFinishedEvent(TypeConverter.fromYarn(t.jobId), 0, 0, 0, 0,
              0, 0, 0, new Counters(), new Counters(), new Counters()), currentTime));
      jheh.getDispatcher().await();
      entities = ts.getEntities("MAPREDUCE_JOB", null, null, null,
              null, null, null, null, null, null);
      Assert.assertEquals(1, entities.getEntities().size());
      tEntity = entities.getEntities().get(0);
      Assert.assertEquals(t.jobId.toString(), tEntity.getEntityId());
      Assert.assertEquals(4, tEntity.getEvents().size());
      Assert.assertEquals(EventType.JOB_SUBMITTED.toString(),
              tEntity.getEvents().get(0).getEventType());
      Assert.assertEquals(EventType.JOB_FINISHED.toString(),
              tEntity.getEvents().get(1).getEventType());
      Assert.assertEquals(EventType.AM_STARTED.toString(),
              tEntity.getEvents().get(2).getEventType());
      Assert.assertEquals(EventType.JOB_QUEUE_CHANGED.toString(),
              tEntity.getEvents().get(3).getEventType());
      Assert.assertEquals(currentTime + 10,
              tEntity.getEvents().get(0).getTimestamp());
      Assert.assertEquals(currentTime,
              tEntity.getEvents().get(1).getTimestamp());
      Assert.assertEquals(currentTime - 10,
              tEntity.getEvents().get(2).getTimestamp());
      Assert.assertEquals(currentTime - 20,
              tEntity.getEvents().get(3).getTimestamp());

      handleEvent(jheh, new JobHistoryEvent(t.jobId,
            new JobUnsuccessfulCompletionEvent(TypeConverter.fromYarn(t.jobId),
            0, 0, 0, 0, 0, 0, 0, JobStateInternal.KILLED.toString()),
            currentTime + 20));
      jheh.getDispatcher().await();
      entities = ts.getEntities("MAPREDUCE_JOB", null, null, null,
              null, null, null, null, null, null);
      Assert.assertEquals(1, entities.getEntities().size());
      tEntity = entities.getEntities().get(0);
      Assert.assertEquals(t.jobId.toString(), tEntity.getEntityId());
      Assert.assertEquals(5, tEntity.getEvents().size());
      Assert.assertEquals(EventType.JOB_KILLED.toString(),
              tEntity.getEvents().get(0).getEventType());
      Assert.assertEquals(EventType.JOB_SUBMITTED.toString(),
              tEntity.getEvents().get(1).getEventType());
      Assert.assertEquals(EventType.JOB_FINISHED.toString(),
              tEntity.getEvents().get(2).getEventType());
      Assert.assertEquals(EventType.AM_STARTED.toString(),
              tEntity.getEvents().get(3).getEventType());
      Assert.assertEquals(EventType.JOB_QUEUE_CHANGED.toString(),
              tEntity.getEvents().get(4).getEventType());
      Assert.assertEquals(currentTime + 20,
              tEntity.getEvents().get(0).getTimestamp());
      Assert.assertEquals(currentTime + 10,
              tEntity.getEvents().get(1).getTimestamp());
      Assert.assertEquals(currentTime,
              tEntity.getEvents().get(2).getTimestamp());
      Assert.assertEquals(currentTime - 10,
              tEntity.getEvents().get(3).getTimestamp());
      Assert.assertEquals(currentTime - 20,
              tEntity.getEvents().get(4).getTimestamp());

      handleEvent(jheh, new JobHistoryEvent(t.jobId,
            new TaskStartedEvent(t.taskID, 0, TaskType.MAP, "")));
      jheh.getDispatcher().await();
      entities = ts.getEntities("MAPREDUCE_TASK", null, null, null,
              null, null, null, null, null, null);
      Assert.assertEquals(1, entities.getEntities().size());
      tEntity = entities.getEntities().get(0);
      Assert.assertEquals(t.taskID.toString(), tEntity.getEntityId());
      Assert.assertEquals(1, tEntity.getEvents().size());
      Assert.assertEquals(EventType.TASK_STARTED.toString(),
              tEntity.getEvents().get(0).getEventType());
      Assert.assertEquals(TaskType.MAP.toString(),
              tEntity.getEvents().get(0).getEventInfo().get("TASK_TYPE"));

      handleEvent(jheh, new JobHistoryEvent(t.jobId,
            new TaskStartedEvent(t.taskID, 0, TaskType.REDUCE, "")));
      jheh.getDispatcher().await();
      entities = ts.getEntities("MAPREDUCE_TASK", null, null, null,
              null, null, null, null, null, null);
      Assert.assertEquals(1, entities.getEntities().size());
      tEntity = entities.getEntities().get(0);
      Assert.assertEquals(t.taskID.toString(), tEntity.getEntityId());
      Assert.assertEquals(2, tEntity.getEvents().size());
      Assert.assertEquals(EventType.TASK_STARTED.toString(),
              tEntity.getEvents().get(1).getEventType());
      Assert.assertEquals(TaskType.REDUCE.toString(),
              tEntity.getEvents().get(0).getEventInfo().get("TASK_TYPE"));
      Assert.assertEquals(TaskType.MAP.toString(),
              tEntity.getEvents().get(1).getEventInfo().get("TASK_TYPE"));
    }
  }