consumer.java

public void consume(Message<String> message) {
    try {
      final EventHandler handler = subscriptionMap.get(event); // Selecting a appropriate handler based on the message
      processWithRetry(handler, message);
    } catch (RequeableException e) {
      failureHandler.reQueue(message, e);
    } catch (DroppableException e) {
      failureHandler.dropMessage(message, e);
    } catch (Exception e) {
      failureHandler.putOnDlq(message, e);
    }
  }

  private void processWithRetry(EventHandler handler, Message<String> message)
      throws Exception {
    retryTemplate.execute(context -> {
      handler.handle(message); // Processing delegated to the handler method
      return null;
    }, context -> {
      final Exception originalError = (Exception) context.getLastThrowable();
      if (originalError instanceof RetryableException) {
        throw new RequeableException(originalError); // After cutoff RetryableException exception becomes RequeableException
      }
      throw originalError;
    });
  }