@Test (timeout=10000)
public void testGetMapCompletionEvents() throws IOException {

  TaskAttemptCompletionEvent[] empty = {};
  TaskAttemptCompletionEvent[] taskEvents = {
      createTce(0, true, TaskAttemptCompletionEventStatus.OBSOLETE),
      createTce(1, false, TaskAttemptCompletionEventStatus.FAILED),
      createTce(2, true, TaskAttemptCompletionEventStatus.SUCCEEDED),
      createTce(3, false, TaskAttemptCompletionEventStatus.FAILED) };
  TaskAttemptCompletionEvent[] mapEvents = { taskEvents[0], taskEvents[2] };
  Job mockJob = mock(Job.class);
  when(mockJob.getTaskAttemptCompletionEvents(0, 100))
    .thenReturn(taskEvents);
  when(mockJob.getTaskAttemptCompletionEvents(0, 2))
    .thenReturn(Arrays.copyOfRange(taskEvents, 0, 2));
  when(mockJob.getTaskAttemptCompletionEvents(2, 100))
    .thenReturn(Arrays.copyOfRange(taskEvents, 2, 4));
  when(mockJob.getMapAttemptCompletionEvents(0, 100)).thenReturn(
      TypeConverter.fromYarn(mapEvents));
  when(mockJob.getMapAttemptCompletionEvents(0, 2)).thenReturn(
      TypeConverter.fromYarn(mapEvents));
  when(mockJob.getMapAttemptCompletionEvents(2, 100)).thenReturn(
      TypeConverter.fromYarn(empty));
  AppContext appCtx = mock(AppContext.class);
  when(appCtx.getJob(any(JobId.class))).thenReturn(mockJob);
  JobTokenSecretManager secret = mock(JobTokenSecretManager.class);
  RMHeartbeatHandler rmHeartbeatHandler =
      mock(RMHeartbeatHandler.class);
  final TaskHeartbeatHandler hbHandler = mock(TaskHeartbeatHandler.class);
  Dispatcher dispatcher = mock(Dispatcher.class);
  EventHandler ea = mock(EventHandler.class);
  when(dispatcher.getEventHandler()).thenReturn(ea);
  when(appCtx.getEventHandler()).thenReturn(ea);
  CheckpointAMPreemptionPolicy policy = new CheckpointAMPreemptionPolicy();
  policy.init(appCtx);
  TaskAttemptListenerImpl listener = new TaskAttemptListenerImpl(
      appCtx, secret, rmHeartbeatHandler, policy) {
    @Override
    protected void registerHeartbeatHandler(Configuration conf) {
      taskHeartbeatHandler = hbHandler;
    }
  };
  Configuration conf = new Configuration();
  listener.init(conf);
  listener.start();
  JobID jid = new JobID("12345", 1);
  TaskAttemptID tid = new TaskAttemptID("12345", 1, TaskType.REDUCE, 1, 0);
  MapTaskCompletionEventsUpdate update =
      listener.getMapCompletionEvents(jid, 0, 100, tid);
  assertEquals(2, update.events.length);
  update = listener.getMapCompletionEvents(jid, 0, 2, tid);
  assertEquals(2, update.events.length);
  update = listener.getMapCompletionEvents(jid, 2, 100, tid);
  assertEquals(0, update.events.length);
}