RetryableRecordReader implements RecordReader<P> { @Override public Record<P> readRecord() throws Exception { return recordReadingTemplate.execute(recordReadingCallable); } RetryableRecordReader(RecordReader<P> delegate, RetryPolicy retryPolicy); @Override void open(); @Override Record<P> readRecord(); @Override void close();  }
@Test(expected = Exception.class) public void readRecord_whenMaxAttemptsExceeded() throws Exception { recordReader = new RetryableRecordReader<>(new UnreliableDataSourceReader(), retryPolicy); recordReader.readRecord(); }
@Test public void readRecord_whenMaxAttemptsNotExceeded() throws Exception { recordReader = new RetryableRecordReader<>(new BetterUnreliableDataSourceReader(), retryPolicy); Record actual = recordReader.readRecord(); assertThat(actual).isEqualTo(record); }