@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 readerShouldNotCacheExceptions() throws IOException {
    ShuffleBatchReader base = mock(ShuffleBatchReader.class);
    CachingShuffleBatchReader reader = new CachingShuffleBatchReader(base);
    when(base.read(null, null)).thenThrow(new IOException("test")).thenReturn(testBatch);
    try {
      reader.read(null, null);
      fail("expected an IOException");
    } catch (Exception e) {
      // Nothing to do -- exception is expected.
    }
    assertThat(reader.read(null, null), equalTo(testBatch));
    verify(base, times(2)).read(null, null);
  }