@Override
  public Reiterator<ShuffleEntry> read(
      @Nullable ShufflePosition startPosition, @Nullable ShufflePosition endPosition) {
    return new ShuffleReadIterator(startPosition, endPosition);
  }
@Test
  public void readerShouldMergeMultipleBatchResults() throws Exception {
    ShuffleEntry e1 = newShuffleEntry(KEY, SKEY, VALUE);
    List<ShuffleEntry> e1s = Collections.singletonList(e1);
    ShuffleEntry e2 = newShuffleEntry(KEY, SKEY, VALUE);
    List<ShuffleEntry> e2s = Collections.singletonList(e2);
    when(batchReader.read(START_POSITION, END_POSITION))
        .thenReturn(new ShuffleBatchReader.Batch(e1s, NEXT_START_POSITION));
    when(batchReader.read(NEXT_START_POSITION, END_POSITION))
        .thenReturn(new ShuffleBatchReader.Batch(e2s, null));
    List<ShuffleEntry> results = newArrayList(reader.read(START_POSITION, END_POSITION));
    assertThat(results, contains(e1, e2));

    verify(batchReader).read(START_POSITION, END_POSITION);
    verify(batchReader).read(NEXT_START_POSITION, END_POSITION);
    verifyNoMoreInteractions(batchReader);
  }