Spring + RabbitMQ Exponential Backoff with RetryTemplate unresponsive

Question

I am working on a Spring project, and am trying to implement exponential backoff with dead lettering for a RabbitMQ queue.
In the process, I've created a Dead Letter Queue and a Dead Letter Exchange (Fanout), and set the x-dead-letter-exchange argument for the original queue to the dead letter exchange's name, and created a RetryTemplate with an ExponentialBackOffPolicy.
For testing purposes, my consumer simply rejects all messages it gets by throwing an exception.

This is what my RabbitMQConfiguration class looks like:

@Configuration
@EnableAutoConfiguration
@PropertySource("file:${HOME}/common/config/wave-planning.properties")
public class RabbitMQConfiguration {

    private final static String QUEUE_NAME = "orderPlanQueue";

    private static final String EXCHANGE_NAME = "orderPlanExchange";

    private static final String DL_EXCHANGE_NAME = "deadLetterExchange";

    private static final String DL_QUEUE_NAME = "deadLetterQueue";

    @Value("${rabbitmq.host:localhost}")
    private String host;

    @Value("${rabbitmq.port:5672}")
    private int port;

    @Value("${rabbitmq.user:guest}")
    private String userName;

    @Value("${rabbitmq.password:guest}")
    private String password;

    @Value("${rabbitmq.initial_backoff_interval:1000}")
    private int INITIAL_INTERVAL_IN_MILLISECONDS;

    @Value("${rabbitmq.max_backoff_interval:10000}")
    private int MAX_INTERVAL_IN_MILLISECONDS;

    @Autowired
    OrderPlanService orderPlanService;

    @Bean
    Queue queue() {
        Map<String, Object> qargs = new HashMap<String, Object>();
        qargs.put("x-dead-letter-exchange", DL_EXCHANGE_NAME);
        return new Queue(QUEUE_NAME, false, false, false, qargs);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    FanoutExchange deadLetterExchange() { return new FanoutExchange(DL_EXCHANGE_NAME); }

    @Bean
    Queue deadLetterQueue() { return new Queue(DL_QUEUE_NAME); }

    @Bean
    Binding deadLetterBinding(Queue deadLetterQueue, FanoutExchange deadLetterExchange) {
        return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange);
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(QUEUE_NAME);
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(userName);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }

    @Bean
    public MessageConverter Jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public AmqpTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());

        RetryTemplate retry = new RetryTemplate();
        ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();

        policy.setInitialInterval(INITIAL_INTERVAL_IN_MILLISECONDS);
        policy.setMultiplier(2);
        policy.setMaxInterval(MAX_INTERVAL_IN_MILLISECONDS);

        retry.setBackOffPolicy(policy);
        template.setRetryTemplate(retry);

        template.setRoutingKey(QUEUE_NAME);
        template.setMessageConverter(Jackson2JsonMessageConverter());
        return template;
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setMessageConverter(Jackson2JsonMessageConverter());
        container.setQueueNames(QUEUE_NAME);
        container.setMessageListener(listenerAdapter);
        container.setDefaultRequeueRejected(false);
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter() {
        return new MessageListenerAdapter(orderPlanService, "consume");
    }
}

The relevant portion of the consumer is basically this:

@Service
@Transactional
public class BaseOrderPlanService implements OrderPlanService {

    ....

    @Override
    public void consume(Object object) {
        throw new IllegalArgumentException("Test");
    }
}

For the autowired integer values, the default value is used.
In running this, I see that the exchanges and queues are created on rabbitmq as expected, with the expected bindings and arguments where relevant.
However, when I pass a message to the orderPlanExchange with routing key "orderPlanQueue", it will cause an infinite loop as the message is rejected and replaced on the queue repeatedly.
If, on the other hand, the IllegalArgumentException is replaced with an AmqpRejectAndDontRequeueException, the message is simply thrown into the dead letter queue on the first rejection attempt.

If anyone could point out what I might be doing wrong here that the retry policy is not being applied, I'd much appreciate it.

Edit: Code with StatefulRetryOperationsInterceptor as per Artem's suggestion.

@Configuration
@EnableAutoConfiguration
@PropertySource("file:${HOME}/common/config/wave-planning.properties")
public class RabbitMQConfiguration {

    private final static String QUEUE_NAME = "orderPlanQueue";

    private static final String EXCHANGE_NAME = "orderPlanExchange";

    private static final String DL_EXCHANGE_NAME = "deadLetterExchange";

    private static final String DL_QUEUE_NAME = "deadLetterQueue";

    @Value("${rabbitmq.host:localhost}")
    private String host;

    @Value("${rabbitmq.port:5672}")
    private int port;

    @Value("${rabbitmq.user:guest}")
    private String userName;

    @Value("${rabbitmq.password:guest}")
    private String password;

    @Value("${rabbitmq.initial_backoff_interval:1000}")
    private int INITIAL_INTERVAL_IN_MILLISECONDS;

    @Value("${rabbitmq.max_backoff_interval:10000}")
    private int MAX_INTERVAL_IN_MILLISECONDS;

    @Autowired
    OrderPlanService orderPlanService;

    @Bean
    Queue queue() {
        Map<String, Object> qargs = new HashMap<String, Object>();
        qargs.put("x-dead-letter-exchange", DL_EXCHANGE_NAME);
        return new Queue(QUEUE_NAME, false, false, false, qargs);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(QUEUE_NAME);
    }

    @Bean
    FanoutExchange deadLetterExchange() { return new FanoutExchange(DL_EXCHANGE_NAME); }

    @Bean
    Queue deadLetterQueue() { return new Queue(DL_QUEUE_NAME); }

    @Bean
    Binding deadLetterBinding(Queue deadLetterQueue, FanoutExchange deadLetterExchange) {
        return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange);
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(userName);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }

    @Bean
    public MessageConverter Jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public AmqpTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());

        /*
        RetryTemplate retry = new RetryTemplate();
        ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();

        policy.setInitialInterval(INITIAL_INTERVAL_IN_MILLISECONDS);
        policy.setMultiplier(2);
        policy.setMaxInterval(MAX_INTERVAL_IN_MILLISECONDS);

        retry.setBackOffPolicy(policy);
        template.setRetryTemplate(retry);
        */

        template.setRoutingKey(QUEUE_NAME);
        template.setMessageConverter(Jackson2JsonMessageConverter());
        return template;
    }

    @Bean
    StatefulRetryOperationsInterceptor interceptor() {
        return RetryInterceptorBuilder.stateful()
                .maxAttempts(4)
                .backOffOptions(INITIAL_INTERVAL_IN_MILLISECONDS, 2, MAX_INTERVAL_IN_MILLISECONDS)
                .build();
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setMessageConverter(Jackson2JsonMessageConverter());
        container.setQueueNames(QUEUE_NAME);
        container.setMessageListener(listenerAdapter);
        container.setAdviceChain(new Advice[] {interceptor()});
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter() {
        return new MessageListenerAdapter(orderPlanService, "consume");
    }

}

Answer 1

The retry policy on RabbitTemplate is fully unrelated to the DLQ/DLX. That is for the consumer side.

See the difference in the Reference Manual here:

you can now configure the RabbitTemplate to use a RetryTemplate to help with handling problems with broker connectivity.

and here:

To put a limit in the client on the number of re-deliveries, one choice is a StatefulRetryOperationsInterceptor in the advice chain of the listener.

So, you have to reconsider your logic and put retry capabilities to the SimpleMessageListenerContainer definition.