protected int getMapCompletionEvents()
      throws IOException, InterruptedException {
    
    int numNewMaps = 0;
    TaskCompletionEvent events[] = null;

    do {
      MapTaskCompletionEventsUpdate update =
          umbilical.getMapCompletionEvents(
              (org.apache.hadoop.mapred.JobID)reduce.getJobID(),
              fromEventIdx,
              maxEventsToFetch,
              (org.apache.hadoop.mapred.TaskAttemptID)reduce);
      events = update.getMapTaskCompletionEvents();
      LOG.debug("Got " + events.length + " map completion events from " +
               fromEventIdx);

      assert !update.shouldReset() : "Unexpected legacy state";

      // Update the last seen event ID
      fromEventIdx += events.length;

      // Process the TaskCompletionEvents:
      // 1. Save the SUCCEEDED maps in knownOutputs to fetch the outputs.
      // 2. Save the OBSOLETE/FAILED/KILLED maps in obsoleteOutputs to stop
      //    fetching from those maps.
      // 3. Remove TIPFAILED maps from neededOutputs since we don't need their
      //    outputs at all.
      for (TaskCompletionEvent event : events) {
        scheduler.resolve(event);
        if (TaskCompletionEvent.Status.SUCCEEDED == event.getTaskStatus()) {
          ++numNewMaps;
        }
      }
    } while (events.length == maxEventsToFetch);

    return numNewMaps;
  }
@Test
  public void testConsecutiveFetch()
      throws IOException, InterruptedException {
    final int MAX_EVENTS_TO_FETCH = 100;
    TaskAttemptID tid = new TaskAttemptID("12345", 1, TaskType.REDUCE, 1, 1);

    TaskUmbilicalProtocol umbilical = mock(TaskUmbilicalProtocol.class);
    when(umbilical.getMapCompletionEvents(any(JobID.class),
        anyInt(), anyInt(), any(TaskAttemptID.class)))
      .thenReturn(getMockedCompletionEventsUpdate(0, 0));
    when(umbilical.getMapCompletionEvents(any(JobID.class),
        eq(0), eq(MAX_EVENTS_TO_FETCH), eq(tid)))
      .thenReturn(getMockedCompletionEventsUpdate(0, MAX_EVENTS_TO_FETCH));
    when(umbilical.getMapCompletionEvents(any(JobID.class),
        eq(MAX_EVENTS_TO_FETCH), eq(MAX_EVENTS_TO_FETCH), eq(tid)))
      .thenReturn(getMockedCompletionEventsUpdate(MAX_EVENTS_TO_FETCH,
          MAX_EVENTS_TO_FETCH));
    when(umbilical.getMapCompletionEvents(any(JobID.class),
        eq(MAX_EVENTS_TO_FETCH*2), eq(MAX_EVENTS_TO_FETCH), eq(tid)))
      .thenReturn(getMockedCompletionEventsUpdate(MAX_EVENTS_TO_FETCH*2, 3));

    @SuppressWarnings("unchecked")
    ShuffleScheduler<String,String> scheduler =
      mock(ShuffleScheduler.class);
    ExceptionReporter reporter = mock(ExceptionReporter.class);

    EventFetcherForTest<String,String> ef =
        new EventFetcherForTest<String,String>(tid, umbilical, scheduler,
            reporter, MAX_EVENTS_TO_FETCH);
    ef.getMapCompletionEvents();

    verify(reporter, never()).reportException(any(Throwable.class));
    InOrder inOrder = inOrder(umbilical);
    inOrder.verify(umbilical).getMapCompletionEvents(any(JobID.class),
        eq(0), eq(MAX_EVENTS_TO_FETCH), eq(tid));
    inOrder.verify(umbilical).getMapCompletionEvents(any(JobID.class),
        eq(MAX_EVENTS_TO_FETCH), eq(MAX_EVENTS_TO_FETCH), eq(tid));
    inOrder.verify(umbilical).getMapCompletionEvents(any(JobID.class),
        eq(MAX_EVENTS_TO_FETCH*2), eq(MAX_EVENTS_TO_FETCH), eq(tid));
    verify(scheduler, times(MAX_EVENTS_TO_FETCH*2 + 3)).resolve(
        any(TaskCompletionEvent.class));
  }