Logging exceptions thrown by message listeners for Spring AMQP

Question

I have an application which handles messages using an async consumer (via @RabbitListener). Inside this consumer method, an exception occurs and the message is requeued due to the policies I've defined:

spring:
    rabbitmq:
        listener:
            simple:
                default-requeue-rejected: false
                retry:
                    enabled: true
                    max-attempts: 10
                    initial-interval: 60000 # a minute
                    multiplier: 2
                    max-interval: 600000 # 10 minutes

The consumer method calls a private method which recursively fetches data from the DB and pushes into a queue using RabbitTemplate. I expect around 200 messages in this queue however it goes up to around 700k and then the consumer thread stops due to the retry policy exhaustion.

The problem is that I cannot find any place to log the exception and therefore I cannot understand which part of the business logic causes this issue. I may try placing the whole function into a try/catch block and log the issue before rethrowing it for Spring AMQP's exception handling but I want to know whether a better approach exists.

My project has the following dependencies:

Spring Boot: 1.5.9.RELEASE
Spring AMQP: 1.7.4.RELEASE
RabbitMQ: 3.7.2

Answer 1

We should probably try to make it a bit easier to add a RetryListener, but you can do it now, by replacing the retry interceptor as follows...

@SpringBootApplication
public class So48331502Application {

    private static final Logger logger = LoggerFactory.getLogger(So48331502Application.class);

    public static void main(String[] args) {
        SpringApplication.run(So48331502Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(RabbitListenerEndpointRegistry registry,
            RabbitProperties properties, Advice interceptor) {
        return args -> {
            ListenerRetry retry = properties.getListener().getSimple().getRetry();
            if (retry.isEnabled()) {
                SimpleMessageListenerContainer container = (SimpleMessageListenerContainer) registry
                        .getListenerContainer("myListener");
                container.setAdviceChain(interceptor);
            }
            registry.start();
        };
    }

    @Bean
    public StatelessRetryOperationsInterceptorFactoryBean interceptor(RabbitProperties properties) {
        ListenerRetry retry = properties.getListener().getSimple().getRetry();
        RetryTemplate retryTemplate = new RetryTemplate();
        RetryPolicy retryPolicy = new SimpleRetryPolicy(retry.getMaxAttempts());
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(retry.getInitialInterval());
        backOffPolicy.setMultiplier(retry.getMultiplier());
        backOffPolicy.setMaxInterval(retry.getMaxInterval());
        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.registerListener(
            new RetryListener() {

                @Override
                public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
                    return true;
                }

                @Override
                public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
                        Throwable throwable) {
                    if (throwable != null) {
                        logger.info("Failed: Retry count " + context.getRetryCount(), throwable);
                    }
                }

                @Override
                public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
                        Throwable throwable) {
                    logger.info("Retry count " + context.getRetryCount(), throwable);
                }
            });
        StatelessRetryOperationsInterceptorFactoryBean interceptor =
                new StatelessRetryOperationsInterceptorFactoryBean();
        interceptor.setRetryOperations(retryTemplate);
        return interceptor;
    }

    @RabbitListener(id="myListener", queues = "one")
    public void in(Object in) {
        throw new RuntimeException();
    }

}

Note that you have to set auto-startup to false so you can change the advice chain...

spring:
  rabbitmq:
    listener:
      simple:
        auto-startup: 'false'
        retry:
          enabled: 'true'

then start the registry, which will start all the containers.