RetryableRecordWriter implements RecordWriter<P> { @Override public void writeRecords(Batch<P> batch) throws Exception { recordWritingTemplate.execute(new RecordWritingCallable(delegate, batch)); } RetryableRecordWriter(RecordWriter<P> delegate, RetryPolicy retryPolicy); @Override void open(); @Override void writeRecords(Batch<P> batch); @Override void close();  }
@Test(expected = Exception.class) public void writeRecords_whenMaxAttemptsExceeded() throws Exception { recordWriter = new RetryableRecordWriter<>(new UnreliableDataSinkWriter(), retryPolicy); recordWriter.writeRecords(new Batch<>(record)); }
@Test public void writeRecords_whenMaxAttemptsNotExceeded() throws Exception { BetterUnreliableDataSinkWriter delegate = new BetterUnreliableDataSinkWriter(); recordWriter = new RetryableRecordWriter<>(delegate, retryPolicy); recordWriter.writeRecords(new Batch<>(record)); assertThat(delegate.isExecuted()).isTrue(); }