Testing

Testing Dead Letter Queues: Kafka, SQS, and RabbitMQ DLQ Strategies

HelpMeTest

HelpMeTest

— 5 min read
Testing Dead Letter Queues: Kafka, SQS, and RabbitMQ DLQ Strategies

Dead letter queues (DLQs) are the safety net of event-driven systems. When a message can't be processed after N retries, it goes to the DLQ for manual inspection or alternative processing. Most teams test the happy path but skip testing DLQ behavior — which means they discover bugs when messages silently disappear in production.

What DLQ Testing Actually Covers

  1. Routing: does a poison message end up in the DLQ after the correct number of retries?
  2. Retry count: is the retry limit enforced correctly?
  3. Message preservation: does the DLQ message contain the original payload and error details?
  4. DLQ consumer: does your DLQ processor handle messages correctly?
  5. Alerting: does the team get notified when messages land in the DLQ?

Testing Kafka DLQ with Spring Kafka

Spring Kafka's DeadLetterPublishingRecoverer routes failed messages to a DLQ topic.

// Configuration
@Configuration
public class KafkaConfig {
    
    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
            template,
            (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
        );
        
        // 3 retries with 1s, 2s, 4s backoff
        ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(1000);
        backOff.setMultiplier(2.0);
        
        return new DefaultErrorHandler(recoverer, backOff);
    }
}
// Integration test
@SpringBootTest
@EmbeddedKafka(topics = {"user-events", "user-events.DLT"})
class DeadLetterQueueIntegrationTest {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    
    @MockBean
    private UserEventProcessor eventProcessor;
    
    @Test
    void poisonMessageRoutesToDLTAfterRetries() throws Exception {
        // Arrange: make the processor always fail
        doThrow(new RuntimeException("Processing failed"))
            .when(eventProcessor).process(any());
        
        // Track messages received on DLT
        List<ConsumerRecord<String, String>> dltMessages = new ArrayList<>();
        CountDownLatch latch = new CountDownLatch(1);
        
        setupDltConsumer(dltMessages, latch);
        
        // Act: send a message that will fail
        kafkaTemplate.send("user-events", "u1", "{\"user_id\":\"u1\",\"event\":\"broken\"}");
        
        // Wait for DLT message (up to 30s for retries + backoff)
        assertTrue(latch.await(30, TimeUnit.SECONDS), "Message did not arrive in DLT");
        
        // Assert
        assertFalse(dltMessages.isEmpty());
        
        ConsumerRecord<String, String> dltRecord = dltMessages.get(0);
        
        // Original payload preserved
        assertEquals("{\"user_id\":\"u1\",\"event\":\"broken\"}", dltRecord.value());
        
        // Error headers added by Spring Kafka
        Header exceptionHeader = dltRecord.headers().lastHeader("kafka_dlt-exception-message");
        assertNotNull(exceptionHeader, "DLT record should have exception header");
        assertThat(new String(exceptionHeader.value())).contains("Processing failed");
        
        // Processor was called 4 times (1 original + 3 retries)
        verify(eventProcessor, times(4)).process(any());
    }
    
    @Test
    void successfulMessageDoesNotRouteToDLT() throws Exception {
        doNothing().when(eventProcessor).process(any());
        
        List<ConsumerRecord<String, String>> dltMessages = new ArrayList<>();
        CountDownLatch latch = new CountDownLatch(1);
        setupDltConsumer(dltMessages, latch);
        
        kafkaTemplate.send("user-events", "u1", "{\"user_id\":\"u1\",\"event\":\"valid\"}");
        
        // DLT should receive nothing
        boolean receivedDlt = latch.await(5, TimeUnit.SECONDS);
        assertFalse(receivedDlt, "Successful message should not go to DLT");
        assertTrue(dltMessages.isEmpty());
    }
    
    private void setupDltConsumer(List<ConsumerRecord<String, String>> messages, CountDownLatch latch) {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
            "dlt-test-consumer", "true", embeddedKafka
        );
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(List.of("user-events.DLT"));
            
            new Thread(() -> {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    records.forEach(r -> {
                        messages.add(r);
                        latch.countDown();
                    });
                }
            }).start();
        }
    }
}

AWS SQS DLQ Testing

# Using moto to mock AWS SQS
import boto3
import pytest
from moto import mock_sqs
import json

@mock_sqs
def test_message_routes_to_dlq_after_max_receive_count():
    sqs = boto3.client('sqs', region_name='us-east-1')
    
    # Create DLQ
    dlq_response = sqs.create_queue(QueueName='user-events-dlq')
    dlq_url = dlq_response['QueueUrl']
    dlq_arn = sqs.get_queue_attributes(
        QueueUrl=dlq_url,
        AttributeNames=['QueueArn']
    )['Attributes']['QueueArn']
    
    # Create main queue with DLQ configured (maxReceiveCount=3)
    queue_response = sqs.create_queue(
        QueueName='user-events',
        Attributes={
            'RedrivePolicy': json.dumps({
                'deadLetterTargetArn': dlq_arn,
                'maxReceiveCount': '3'
            }),
            'VisibilityTimeout': '1'  # Short timeout for testing
        }
    )
    queue_url = queue_response['QueueUrl']
    
    # Send a message
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({'user_id': 'u1', 'event': 'USER_CREATED'})
    )
    
    # Receive and "fail" the message 3 times (don't delete it)
    for attempt in range(3):
        messages = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=1,
            VisibilityTimeout=1
        ).get('Messages', [])
        
        if messages:
            # Simulate processing failure: message stays in queue
            # In real processing, you'd NOT call delete_message on failure
            pass
        
        # Wait for visibility timeout to expire
        import time
        time.sleep(1.1)
    
    # After maxReceiveCount exhausted, SQS routes to DLQ
    # NOTE: moto simulates this behavior
    dlq_messages = sqs.receive_message(
        QueueUrl=dlq_url,
        MaxNumberOfMessages=1
    ).get('Messages', [])
    
    # In real AWS, this would have the original message
    # moto may not fully simulate automatic DLQ routing — use real SQS in integration tests
    assert True  # Integration test marker

@mock_sqs
def test_dlq_processor_handles_malformed_message():
    """Test the DLQ consumer itself"""
    sqs = boto3.client('sqs', region_name='us-east-1')
    
    dlq_response = sqs.create_queue(QueueName='user-events-dlq')
    dlq_url = dlq_response['QueueUrl']
    
    # Simulate a malformed message landing in DLQ
    sqs.send_message(
        QueueUrl=dlq_url,
        MessageBody='{"corrupted": true, "original_error": "JSON parse failed"}',
        MessageAttributes={
            'original_topic': {
                'DataType': 'String',
                'StringValue': 'user-events'
            }
        }
    )
    
    # Test the DLQ processor
    processor = DlqProcessor(sqs_client=sqs, alert_service=MockAlertService())
    result = processor.process_dlq(dlq_url, max_messages=10)
    
    assert result.processed_count == 1
    assert result.alert_sent == True  # DLQ messages should trigger alerts

RabbitMQ DLQ Testing

# Testing RabbitMQ DLQ with pika and a real broker via Testcontainers
import pytest
import pika
import time
from testcontainers.rabbitmq import RabbitMqContainer

@pytest.fixture(scope='module')
def rabbitmq():
    with RabbitMqContainer("rabbitmq:3.12-management") as rabbit:
        yield rabbit.get_connection_params()

def test_rejected_message_routes_to_dlq(rabbitmq):
    connection = pika.BlockingConnection(rabbitmq)
    channel = connection.channel()
    
    # Set up DLQ exchange and queue
    channel.exchange_declare(exchange='dlx', exchange_type='direct')
    channel.queue_declare(queue='user-events-dlq')
    channel.queue_bind(queue='user-events-dlq', exchange='dlx', routing_key='user-events')
    
    # Set up main queue with DLX configured
    channel.queue_declare(
        queue='user-events',
        arguments={
            'x-dead-letter-exchange': 'dlx',
            'x-dead-letter-routing-key': 'user-events',
            'x-message-ttl': 1000  # 1s TTL for testing
        }
    )
    
    # Publish a message
    channel.basic_publish(
        exchange='',
        routing_key='user-events',
        body=b'{"user_id": "u1", "event": "broken"}',
        properties=pika.BasicProperties(delivery_mode=2)  # persistent
    )
    
    # Consume and reject the message (simulating processing failure)
    received = []
    
    def on_message(ch, method, properties, body):
        received.append(body)
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)  # reject, don't requeue
    
    channel.basic_consume(queue='user-events', on_message_callback=on_message)
    connection.process_data_events(time_limit=2)  # process for 2s
    
    # Give RabbitMQ time to route to DLQ
    time.sleep(0.5)
    
    # Check DLQ
    method, properties, body = channel.basic_get('user-events-dlq', auto_ack=True)
    
    assert body is not None, "Message should be in DLQ after rejection"
    assert b'"user_id": "u1"' in body
    
    connection.close()

Testing DLQ Monitoring and Alerting

The DLQ is only useful if someone acts on it. Test that your monitoring fires.

def test_dlq_depth_triggers_alert():
    mock_sqs = MagicMock()
    mock_alert = MagicMock()
    
    # Simulate DLQ with 5 messages
    mock_sqs.get_queue_attributes.return_value = {
        'Attributes': {'ApproximateNumberOfMessages': '5'}
    }
    
    monitor = DlqMonitor(
        sqs_client=mock_sqs,
        alert_service=mock_alert,
        threshold=3  # alert if > 3 messages in DLQ
    )
    
    monitor.check_dlq('https://sqs.us-east-1.amazonaws.com/123/my-dlq')
    
    mock_alert.send.assert_called_once_with(
        channel='#alerts-critical',
        message=pytest.approx(
            'DLQ my-dlq has 5 messages (threshold: 3)',
            abs=False
        )
    )

def test_dlq_below_threshold_does_not_alert():
    mock_sqs = MagicMock()
    mock_alert = MagicMock()
    
    mock_sqs.get_queue_attributes.return_value = {
        'Attributes': {'ApproximateNumberOfMessages': '2'}
    }
    
    monitor = DlqMonitor(sqs_client=mock_sqs, alert_service=mock_alert, threshold=3)
    monitor.check_dlq('https://sqs.us-east-1.amazonaws.com/123/my-dlq')
    
    mock_alert.send.assert_not_called()

The DLQ Testing Checklist

Before going to production with any message-driven system:

  • Poison messages route to DLQ after the correct retry count
  • Retry count is enforced (not more, not less)
  • Original message payload is preserved in DLQ
  • Error details (exception type, message) are stored as headers/attributes
  • DLQ consumer processes messages correctly
  • DLQ depth monitoring and alerting is tested
  • Re-processing from DLQ works (when you fix the bug and want to replay)

The DLQ is your last line of defense. Test it like it matters — because when production breaks at 3am, it does.

Read more

Start now free