Testing

RabbitMQ Spring AMQP Testing: DLQ, Acknowledgements, and Retry Patterns

HelpMeTest

HelpMeTest

— 6 min read
RabbitMQ Spring AMQP Testing: DLQ, Acknowledgements, and Retry Patterns

Spring AMQP applications need more than basic consumer tests — dead-letter queues, acknowledgment modes, retry policies, and poison pill rejection all require explicit test coverage. Testcontainers provides a real RabbitMQ instance; Spring's RabbitAdmin and RabbitTemplate let you query queue state and send test messages. The hardest scenarios (nack, DLQ routing, retry exhaustion) are also the most important to verify.

Key Takeaways

Test DLQ routing explicitly. Configure your topology — exchange, binding, dead-letter exchange, DLQ — in your test setup, then simulate processing failures and assert messages appear on the DLQ.

Use channel.basicNack with requeue=false to trigger DLQ routing. A nacked message with requeue=false goes to the dead-letter exchange; with requeue=true it goes back to the front of the queue.

Assert acknowledgment mode with AcknowledgeMode.MANUAL. With AUTO mode the container acks immediately. With MANUAL you control exactly when the ack fires — test that it fires only after successful processing.

Test retry exhaustion. Configure RetryOperationsInterceptor with a max-attempts limit. After N failures the framework delivers to the DLQ. Verify both the retry count and the final DLQ delivery.

Use RabbitAdmin.getQueueInfo() to assert queue depth. After a test action, queueInfo.getMessageCount() tells you how many messages are waiting — avoid sleeping by polling with Awaitility.

What Spring AMQP Tests Must Cover

A basic happy-path consumer test — send a message, assert it was processed — leaves the most dangerous failure modes untested. Production issues in RabbitMQ applications cluster around:

  • Messages that fail processing and loop back into the queue indefinitely
  • Dead-letter queues that never receive messages because the binding is misconfigured
  • Acknowledgment bugs that cause messages to be lost or reprocessed after restart
  • Retry policies that retry forever instead of eventually rejecting

Each of these requires an explicit test.

Test Setup with Testcontainers

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>rabbitmq</artifactId>
    <scope>test</scope>
</dependency>
@SpringBootTest
@Testcontainers
class RabbitMQAdvancedTest {

    @Container
    static RabbitMQContainer rabbit =
        new RabbitMQContainer(DockerImageName.parse("rabbitmq:3.13-management"))
            .withVhost("test-vhost")
            .withUser("testuser", "testpass");

    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.rabbitmq.host", rabbit::getHost);
        registry.add("spring.rabbitmq.port", rabbit::getAmqpPort);
        registry.add("spring.rabbitmq.username", () -> "testuser");
        registry.add("spring.rabbitmq.password", () -> "testpass");
        registry.add("spring.rabbitmq.virtual-host", () -> "test-vhost");
    }

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitAdmin rabbitAdmin;
}

Testing Dead-Letter Queue Configuration

The most common RabbitMQ bug in production: the DLQ exists but messages never reach it because the dead-letter exchange binding is wrong. Test it explicitly:

@Configuration
class OrderQueueConfig {

    public static final String ORDERS_EXCHANGE = "orders.exchange";
    public static final String ORDERS_QUEUE = "orders.queue";
    public static final String ORDERS_DLX = "orders.dlx";
    public static final String ORDERS_DLQ = "orders.dlq";

    @Bean
    DirectExchange ordersExchange() {
        return new DirectExchange(ORDERS_EXCHANGE);
    }

    @Bean
    DirectExchange ordersDlx() {
        return new DirectExchange(ORDERS_DLX);
    }

    @Bean
    Queue ordersQueue() {
        return QueueBuilder.durable(ORDERS_QUEUE)
            .deadLetterExchange(ORDERS_DLX)
            .deadLetterRoutingKey(ORDERS_DLQ)
            .build();
    }

    @Bean
    Queue ordersDlq() {
        return QueueBuilder.durable(ORDERS_DLQ).build();
    }

    @Bean
    Binding ordersBinding() {
        return BindingBuilder.bind(ordersQueue())
            .to(ordersExchange())
            .with("order.created");
    }

    @Bean
    Binding dlqBinding() {
        return BindingBuilder.bind(ordersDlq())
            .to(ordersDlx())
            .with(ORDERS_DLQ);
    }
}
@Test
void nackedMessageRoutesToDLQ() throws Exception {
    // Send a message that the consumer will reject
    rabbitTemplate.convertAndSend(ORDERS_EXCHANGE, "order.created",
        new OrderMessage("ORDER-POISON", null, BigDecimal.ZERO));  // null customerId = validation failure

    // Wait for the message to be processed and routed to DLQ
    await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
        QueueInformation dlqInfo = rabbitAdmin.getQueueInfo(ORDERS_DLQ);
        assertNotNull(dlqInfo);
        assertEquals(1, dlqInfo.getMessageCount());
    });

    // Verify the original queue is now empty
    QueueInformation queueInfo = rabbitAdmin.getQueueInfo(ORDERS_QUEUE);
    assertEquals(0, queueInfo.getMessageCount());

    // Read from DLQ and assert the message is there with death headers
    Message dlqMessage = rabbitTemplate.receive(ORDERS_DLQ, 5000);
    assertNotNull(dlqMessage);
    assertEquals("ORDER-POISON",
        ((OrderMessage) rabbitTemplate.getMessageConverter()
            .fromMessage(dlqMessage)).getOrderId());
}

Testing Manual Acknowledgment

With AcknowledgeMode.MANUAL, your consumer controls exactly when RabbitMQ considers the message delivered. A bug where you ack before processing means data loss on crashes.

@Component
class OrderConsumer {

    private final OrderService orderService;

    @RabbitListener(queues = ORDERS_QUEUE,
                    ackMode = "MANUAL",
                    containerFactory = "manualAckContainerFactory")
    public void consume(OrderMessage order, Channel channel,
                        @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        try {
            orderService.processOrder(order);
            channel.basicAck(deliveryTag, false);  // ack AFTER processing
        } catch (ValidationException e) {
            channel.basicNack(deliveryTag, false, false);  // nack, don't requeue → DLQ
        } catch (TransientException e) {
            channel.basicNack(deliveryTag, false, true);   // nack, requeue → retry
        }
    }
}
@Test
void acksMessageOnlyAfterSuccessfulProcessing() throws Exception {
    AtomicBoolean processingStarted = new AtomicBoolean(false);
    AtomicBoolean processingComplete = new AtomicBoolean(false);

    // Use a spy to intercept processing
    doAnswer(invocation -> {
        processingStarted.set(true);
        Thread.sleep(200);  // simulate work
        processingComplete.set(true);
        return null;
    }).when(orderService).processOrder(any());

    rabbitTemplate.convertAndSend(ORDERS_EXCHANGE, "order.created",
        new OrderMessage("ORDER-001", "CUST-001", BigDecimal.valueOf(99.99)));

    // Wait for processing to start
    await().atMost(5, TimeUnit.SECONDS).until(processingStarted::get);

    // While processing is happening, the message should still be in-flight (unacked)
    QueueInformation queueInfo = rabbitAdmin.getQueueInfo(ORDERS_QUEUE);
    // unackedMessages > 0 while processing is in progress
    // (this verifies the ack hasn't fired prematurely)

    // Wait for processing to complete
    await().atMost(5, TimeUnit.SECONDS).until(processingComplete::get);

    // After processing, queue should be empty (acked)
    await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
        QueueInformation info = rabbitAdmin.getQueueInfo(ORDERS_QUEUE);
        assertEquals(0, info.getMessageCount());
    });
}

@Test
void nackWithoutRequeueRoutesToDLQ() throws Exception {
    // Send a message with invalid data that will be rejected
    rabbitTemplate.convertAndSend(ORDERS_EXCHANGE, "order.created",
        new OrderMessage("BAD-ORDER", null, BigDecimal.valueOf(-1)));  // negative amount

    // Should appear in DLQ, not be requeued
    await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
        QueueInformation dlqInfo = rabbitAdmin.getQueueInfo(ORDERS_DLQ);
        assertEquals(1, dlqInfo.getMessageCount());
        QueueInformation queueInfo = rabbitAdmin.getQueueInfo(ORDERS_QUEUE);
        assertEquals(0, queueInfo.getMessageCount());
    });
}

Testing Retry Exhaustion

Spring AMQP's retry interceptor retries failed messages before routing to the DLQ. Test that it retries the correct number of times and then gives up:

@Configuration
class RetryConfig {

    @Bean
    SimpleRabbitListenerContainerFactory retryContainerFactory(
            ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);

        RetryTemplate retryTemplate = RetryTemplate.builder()
            .maxAttempts(3)
            .fixedBackoff(100)  // 100ms between retries
            .retryOn(TransientException.class)
            .build();

        factory.setAdviceChain(RetryInterceptorBuilder.stateless()
            .retryOperations(retryTemplate)
            .recoverer(new RejectAndDontRequeueRecoverer())  // goes to DLQ after exhaustion
            .build());

        return factory;
    }
}
@Test
void retries3TimesBeforeRoutingToDLQ() throws Exception {
    AtomicInteger attemptCount = new AtomicInteger(0);

    doAnswer(invocation -> {
        attemptCount.incrementAndGet();
        throw new TransientException("Simulated transient failure");
    }).when(orderService).processOrder(any());

    rabbitTemplate.convertAndSend(ORDERS_EXCHANGE, "order.created",
        new OrderMessage("RETRY-ORDER", "CUST-001", BigDecimal.TEN));

    // After retry exhaustion, message lands in DLQ
    await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> {
        QueueInformation dlqInfo = rabbitAdmin.getQueueInfo(ORDERS_DLQ);
        assertEquals(1, dlqInfo.getMessageCount());
    });

    // Verify exactly 3 attempts were made (not 1, not infinite)
    assertEquals(3, attemptCount.get());
}

Testing Message Ordering and Priority

If your consumers rely on message ordering, test it explicitly — RabbitMQ doesn't guarantee order across multiple consumers on the same queue.

@Test
void processesPriorityMessagesFirst() throws Exception {
    // Use a priority queue
    Queue priorityQueue = QueueBuilder.durable("priority.queue")
        .maxPriority(10)
        .build();
    rabbitAdmin.declareQueue(priorityQueue);

    List<String> processedOrder = Collections.synchronizedList(new ArrayList<>());

    // Send low-priority message first
    MessageProperties lowProps = new MessageProperties();
    lowProps.setPriority(1);
    rabbitTemplate.send("priority.queue",
        new Message("low-priority".getBytes(), lowProps));

    // Then high-priority
    MessageProperties highProps = new MessageProperties();
    highProps.setPriority(9);
    rabbitTemplate.send("priority.queue",
        new Message("high-priority".getBytes(), highProps));

    // Consumer with a single thread should process high-priority first
    // (requires maxConcurrentConsumers=1 so both messages queue before any consumption)
    await().atMost(5, TimeUnit.SECONDS)
        .until(() -> processedOrder.size() == 2);

    assertEquals("high-priority", processedOrder.get(0));
    assertEquals("low-priority", processedOrder.get(1));
}

Testing Connection Failure Recovery

@Test
void recoversAfterBrokerRestart() throws Exception {
    // Send initial message
    rabbitTemplate.convertAndSend(ORDERS_EXCHANGE, "order.created",
        new OrderMessage("ORDER-BEFORE", "CUST-001", BigDecimal.TEN));

    await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
        verify(orderService, times(1)).processOrder(any()));

    // Simulate broker restart
    rabbit.stop();
    rabbit.start();

    // Spring AMQP's SimpleMessageListenerContainer auto-recovers
    // Wait for reconnection
    Thread.sleep(2000);

    // Send another message after recovery
    rabbitTemplate.convertAndSend(ORDERS_EXCHANGE, "order.created",
        new OrderMessage("ORDER-AFTER", "CUST-001", BigDecimal.valueOf(50)));

    await().atMost(15, TimeUnit.SECONDS).untilAsserted(() ->
        verify(orderService, times(2)).processOrder(any()));
}

Testing Message Conversion

Test your message converters explicitly — wrong content-type headers or serialization bugs cause silent failures:

@Test
void deserializesJsonMessageCorrectly() {
    String json = """
        {
            "orderId": "ORDER-JSON",
            "customerId": "CUST-001",
            "amount": 149.99,
            "items": ["item-A", "item-B"]
        }
        """;

    MessageProperties props = new MessageProperties();
    props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
    Message message = new Message(json.getBytes(StandardCharsets.UTF_8), props);

    OrderMessage order = (OrderMessage) rabbitTemplate.getMessageConverter()
        .fromMessage(message);

    assertEquals("ORDER-JSON", order.getOrderId());
    assertEquals(2, order.getItems().size());
    assertEquals(BigDecimal.valueOf(149.99), order.getAmount());
}

@Test
void rejectsUnknownContentType() {
    MessageProperties props = new MessageProperties();
    props.setContentType("application/xml");
    Message xmlMessage = new Message("<order/>".getBytes(), props);

    assertThrows(MessageConversionException.class, () ->
        rabbitTemplate.getMessageConverter().fromMessage(xmlMessage));
}

What Not to Test

Broker routing internals. Exchange-to-queue routing is RabbitMQ's job, not yours. Test that your binding configuration is correct (via integration test), but don't unit-test AMQP routing rules.

Spring AMQP's retry mechanism itself. The retry logic is tested by Spring. What you're testing is that your code configures it correctly and that your business logic throws the right exception types.

Network-level timeouts. Simulating TCP-level drops requires network chaos tooling — Testcontainers container restarts are sufficient for connection recovery tests.

Read more