@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"));
  }
}