### Question:
EventFetcher extends Thread { 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"; fromEventIdx += events.length; for (TaskCompletionEvent event : events) { scheduler.resolve(event); if (TaskCompletionEvent.Status.SUCCEEDED == event.getTaskStatus()) { ++numNewMaps; } } } while (events.length == maxEventsToFetch); return numNewMaps; } EventFetcher(TaskAttemptID reduce,
                      TaskUmbilicalProtocol umbilical,
                      ShuffleScheduler<K,V> scheduler,
                      ExceptionReporter reporter,
                      int maxEventsToFetch); @Override void run(); void shutDown();  }### Answer:
@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)); }