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