RetryableRecordProcessor implements RecordProcessor<I, O> { @Override public Record<O> processRecord(Record<I> record) throws Exception { return recordProcessingTemplate.execute(new RecordProcessingCallable<>(delegate, record)); } RetryableRecordProcessor(RecordProcessor<I, O> delegate, RetryPolicy retryPolicy); @Override Record<O> processRecord(Record<I> record);  }
@Test(expected = Exception.class) public void processRecord_whenMaxAttemptsExceeded() throws Exception { recordProcessor = new RetryableRecordProcessor<>(new UnreliableRecordProcessor(), retryPolicy); recordProcessor.processRecord(record); }
@Test public void processRecord_whenMaxAttemptsNotExceeded() throws Exception { BetterUnreliableRecordProcessor delegate = new BetterUnreliableRecordProcessor(); recordProcessor = new RetryableRecordProcessor<>(delegate, retryPolicy); recordProcessor.processRecord(record); assertThat(delegate.isExecuted()).isTrue(); }