### Question:
ShuffleInputEventHandlerImpl implements ShuffleEventHandler { @Override public void handleEvents(List<Event> events) throws IOException { for (Event event : events) { handleEvent(event); } } ShuffleInputEventHandlerImpl(TezInputContext inputContext,
      ShuffleManager shuffleManager,
      FetchedInputAllocator inputAllocator, CompressionCodec codec,
      boolean ifileReadAhead, int ifileReadAheadLength); @Override void handleEvents(List<Event> events);  }### Answer:
@Test public void testSimple() throws IOException { TezInputContext inputContext = mock(TezInputContext.class); ShuffleManager shuffleManager = mock(ShuffleManager.class); FetchedInputAllocator inputAllocator = mock(FetchedInputAllocator.class); ShuffleInputEventHandlerImpl handler = new ShuffleInputEventHandlerImpl(inputContext, shuffleManager, inputAllocator, null, false, 0); int taskIndex = 1; Event dme = createDataMovementEvent(0, taskIndex, null); List<Event> eventList = new LinkedList<Event>(); eventList.add(dme); handler.handleEvents(eventList); InputAttemptIdentifier expectedIdentifier = new InputAttemptIdentifier(taskIndex, 0, PATH_COMPONENT); verify(shuffleManager).addKnownInput(eq(HOST), eq(PORT), eq(expectedIdentifier), eq(0)); }
@Test public void testCurrentPartitionEmpty() throws IOException { TezInputContext inputContext = mock(TezInputContext.class); ShuffleManager shuffleManager = mock(ShuffleManager.class); FetchedInputAllocator inputAllocator = mock(FetchedInputAllocator.class); ShuffleInputEventHandlerImpl handler = new ShuffleInputEventHandlerImpl(inputContext, shuffleManager, inputAllocator, null, false, 0); int taskIndex = 1; Event dme = createDataMovementEvent(0, taskIndex, createEmptyPartitionByteString(0)); List<Event> eventList = new LinkedList<Event>(); eventList.add(dme); handler.handleEvents(eventList); InputAttemptIdentifier expectedIdentifier = new InputAttemptIdentifier(taskIndex, 0); verify(shuffleManager).addCompletedInputWithNoData(eq(expectedIdentifier)); }
@Test public void testOtherPartitionEmpty() throws IOException { TezInputContext inputContext = mock(TezInputContext.class); ShuffleManager shuffleManager = mock(ShuffleManager.class); FetchedInputAllocator inputAllocator = mock(FetchedInputAllocator.class); ShuffleInputEventHandlerImpl handler = new ShuffleInputEventHandlerImpl(inputContext, shuffleManager, inputAllocator, null, false, 0); int taskIndex = 1; Event dme = createDataMovementEvent(0, taskIndex, createEmptyPartitionByteString(1)); List<Event> eventList = new LinkedList<Event>(); eventList.add(dme); handler.handleEvents(eventList); InputAttemptIdentifier expectedIdentifier = new InputAttemptIdentifier(taskIndex, 0, PATH_COMPONENT); verify(shuffleManager).addKnownInput(eq(HOST), eq(PORT), eq(expectedIdentifier), eq(0)); }
@Test public void testMultipleEvents1() throws IOException { TezInputContext inputContext = mock(TezInputContext.class); ShuffleManager shuffleManager = mock(ShuffleManager.class); FetchedInputAllocator inputAllocator = mock(FetchedInputAllocator.class); ShuffleInputEventHandlerImpl handler = new ShuffleInputEventHandlerImpl(inputContext, shuffleManager, inputAllocator, null, false, 0); int taskIndex1 = 1; Event dme1 = createDataMovementEvent(0, taskIndex1, createEmptyPartitionByteString(0)); int taskIndex2 = 2; Event dme2 = createDataMovementEvent(0, taskIndex2, null); List<Event> eventList = new LinkedList<Event>(); eventList.add(dme1); eventList.add(dme2); handler.handleEvents(eventList); InputAttemptIdentifier expectedIdentifier1 = new InputAttemptIdentifier(taskIndex1, 0); InputAttemptIdentifier expectedIdentifier2 = new InputAttemptIdentifier(taskIndex2, 0, PATH_COMPONENT); verify(shuffleManager).addCompletedInputWithNoData(eq(expectedIdentifier1)); verify(shuffleManager).addKnownInput(eq(HOST), eq(PORT), eq(expectedIdentifier2), eq(0)); }