ExponentialBackoffMsgRetryManager implements FailedMsgRetryManager { @Override public void retryStarted(Long offset) { MessageRetryRecord record = this.records.get(offset); if (record == null || !this.waiting.contains(record)) { throw new IllegalStateException("cannot retry a message that has not failed"); } else { this.waiting.remove(record); } } ExponentialBackoffMsgRetryManager(long retryInitialDelayMs, double retryDelayMultiplier, long retryDelayMaxMs); @Override void failed(Long offset); @Override void acked(Long offset); @Override void retryStarted(Long offset); @Override Long nextFailedMessageToRetry(); @Override boolean shouldRetryMsg(Long offset);  }
@Test(expected = IllegalStateException.class) public void testRetryWithoutFail() throws Exception { ExponentialBackoffMsgRetryManager manager = new ExponentialBackoffMsgRetryManager(0, 0d, 0); manager.retryStarted(TEST_OFFSET); }