title: How to wrap exception on exhausted retries with Spring Retry
body: <p><strong>Context:</strong></p>
  
  <p>I'm using spring-retry to retry restTemplate calls.</p>
  
  <p>The restTemplate calls are called from a kafka listener.
  The kafka listener is also configured to retry on error (if any exception are thrown during the process, not only the restTemplate call).</p>
  
  <p><strong>Goal:</strong></p>
  
  <p>I'd like to prevent kafka from retrying when the error come from a retry template which has exhausted.</p>
  
  <p><strong>Actual behavior :</strong></p>
  
  <p>When the retryTemplate exhaust all retries, the original exception is thrown. Thus preventing me from identifying if the error was retried by the retryTemplate.</p>
  
  <p><strong>Desired behavior:</strong></p>
  
  <p>When the retryTemplate exhaust all retries, wrap the original exception in a RetryExhaustedException which will allow me to blacklist it from kafka retries.</p>
  
  <p><strong>Question:</strong></p>
  
  <p>How can I do something like this ?</p>
  
  <p>Thanks</p>
  
  <p><strong>Edit</strong></p>
  
  <p>RetryTemplate configuration :</p>
  
  <pre><code>RetryTemplate retryTemplate = new RetryTemplate();
  
  FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
  backOffPolicy.setBackOffPeriod(1000);
  retryTemplate.setBackOffPolicy(backOffPolicy);
  
  Map&lt;Class&lt;? extends Throwable&gt;, Boolean&gt; retryableExceptions = new HashMap&lt;&gt;();
  retryableExceptions.put(FunctionalException.class, false);
  SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(3, retryableExceptions, true, true);
  retryTemplate.setRetryPolicy(retryPolicy);
  
  retryTemplate.setThrowLastExceptionOnExhausted(false);
  </code></pre>
  
  <p>Kafka ErrorHandler</p>
  
  <pre><code>public class DefaultErrorHandler implements ErrorHandler {
  
  @Override
  public void handle(Exception thrownException, ConsumerRecord&lt;?, ?&gt; data) {
      Throwable exception = Optional.ofNullable(thrownException.getCause()).orElse(thrownException);
  
      // TODO if exception as been retried in a RetryTemplate, stop it to prevent rollback and send it to a DLQ
      // else rethrow exception, it will be rollback and handled by AfterRollbackProcessor to be retried
      throw new KafkaException("Could not handle exception", thrownException);
     }
  }
  </code></pre>
  
  <p>Listener kafka :</p>
  
  <pre><code>@KafkaListener
  public void onMessage(ConsumerRecord&lt;String, String&gt; record) {
      retryTemplate.execute((args) -&gt; {
          throw new RuntimeException("Should be catched by ErrorHandler to prevent rollback");
      }
      throw new RuntimeException("Should be retried by afterRollbackProcessor");
  }
  </code></pre>