@Override
  public Batch read(@Nullable ShufflePosition startPosition, @Nullable ShufflePosition endPosition)
      throws IOException {
    final BatchRange range = new BatchRange(startPosition, endPosition);
    try {
      return cache.get(range);
    } catch (RuntimeException | ExecutionException e) {
      Throwables.propagateIfPossible(e, IOException.class);
      throw new RuntimeException("unexpected", e);
    }
  }
@Test
  public void readerShouldRereadEvictedBatches() throws IOException, ExecutionException {
    ShuffleBatchReader base = mock(ShuffleBatchReader.class);
    CachingShuffleBatchReader reader = new CachingShuffleBatchReader(base);
    when(base.read(null, null)).thenReturn(testBatch);
    ShuffleBatchReader.Batch read = reader.read(null, null);
    assertThat(read, equalTo(testBatch));
    verify(base, times(1)).read(null, null);
    CachingShuffleBatchReader.BatchRange range =
        new CachingShuffleBatchReader.BatchRange(null, null);
    CachingShuffleBatchReader.Batch batch = reader.cache.get(range);
    assertThat(batch, notNullValue());
    reader.cache.invalidateAll();
    read = reader.read(null, null);
    assertThat(read, equalTo(testBatch));
    verify(base, times(2)).read(null, null);
  }