Some Random Notes

4. DLQ

5 min read

Why a Standard DLQ Breaks Ordering (The Simple Explanation)

Think of a Kafka partition like a single-file checkout line at a grocery store. Messages are customers waiting in line, and your consumer is the cashier processing them one by one.

Imagine one customer, John, has a problem with his credit card (a failed message).

If you use a Standard DLQ, it is like pulling John out of the line and putting him in a “troubleshooting room” (the DLQ) to fix his card later, while the cashier immediately moves on to the next person in line.

Here is why that breaks ordering:

  1. Message 1 (Create Order): John buys a shirt. (Success)
  2. Message 2 (Update Order): John changes the shipping address. (Fails! Shunted to the DLQ).
  3. Message 3 (Ship Order): The cashier immediately processes this message. The system ships the shirt to the old address because Message 2 hasn’t happened yet.
  4. DLQ Processing: 10 minutes later, the DLQ fixes Message 2 and updates the address.

Result: The shirt was shipped to the wrong place. Even though you processed all messages, out-of-order execution caused a business failure.

To fix this, you have two choices:

  • Option A: Stop the Line (Spring Kafka’s Default Strategy). You pause the entire partition until John’s issue is resolved. This guarantees order, but no one else behind John can check out.
  • Option B: Entity-based Hold (Smart Routing). You pull John and any future messages belonging to John out of the main line, while allowing other customers (like Mary or Dave) to continue checking out normally.

Here is how both options look in code.

1. Spring Kafka Strategy (Stop the Line / Retry Block)

Spring Kafka manages this out of the box using a DefaultErrorHandler paired with a FixedBackOff or ExponentialBackOff. If a message fails, Spring stops the consumer, rewinds the partition offset, waits a moment, and tries the exact same message again. If it fails repeatedly, it can send it to a DLQ, but it stops the partition while retrying.

The Code (Spring Boot Configuration)

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;
 
@Configuration
public class KafkaConfig {
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
 
        // ─── SPRING KAFKA ORDERED RETRY ERROR HANDLER ────────────────────────
        // This handler catches exceptions thrown by your listener.
        // It will retry the EXACT failed message 3 times, waiting 2 seconds between attempts.
        // Crucially, it PAUSES the partition poll loop during these retries to preserve order.
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(
                new FixedBackOff(2000L, 3) // 2000ms delay, 3 retry attempts
        );
 
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}

The Listener Code

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
 
@Service
public class OrderConsumerService {
 
    @KafkaListener(topics = "order-events", groupId = "spring-ordered-group")
    public void listen(String message) {
        System.out.println("Processing message: " + message);
        
        if (message.contains("FAIL")) {
            // Throwing a runtime exception triggers the Spring ErrorHandler configured above.
            // Spring will halt consumption of this partition and retry this exact message.
            throw new RuntimeException("Simulated processing crash!");
        }
    }
}

2. Entity-Based Hold State (Smart/Advanced Strategy)

If stopping the entire line isn’t acceptable because one user is broken, you can build a Hold State in your application database.

If a message for customer_abc fails, you flag customer_abc as “BLOCKED” in a fast in-memory store or database (like Redis). You send the failed message to a DLQ. For every new message that arrives, you check if that specific key is blocked. If it is, you instantly route it to the DLQ as well, preserving the relative order behind the failed message.

The Pure Java Code

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
 
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
 
public class EntityHoldConsumer {
    // In production, use Redis or a DB instead of an in-memory HashSet
    private static final Set<String> blockedEntities = new HashSet<>();
 
    public static void main(String[] args) {
        Properties consumerProps = new Properties(); // ... Standard properties here
        Properties producerProps = new Properties(); // ... Standard producer properties here
 
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
             KafkaProducer<String, String> dlqProducer = new KafkaProducer<>(producerProps)) {
            
            consumer.subscribe(Collections.singletonList("order-events"));
 
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
 
                for (ConsumerRecord<String, String> record : records) {
                    String entityKey = record.key();
 
                    // Step 1: Check if this specific entity is already blocked
                    if (blockedEntities.contains(entityKey)) {
                        // Order is already broken for this entity, bypass logic and route straight to DLQ
                        System.out.printf("Entity %s is blocked. Redirecting message sequentially to DLQ.%n", entityKey);
                        dlqProducer.send(new ProducerRecord<>("order-events-dlq", entityKey, record.value()));
                        continue;
                    }
 
                    // Step 2: Attempt standard processing
                    try {
                        executeBusinessLogic(record);
                    } catch (Exception e) {
                        System.err.printf("Failed to process entity %s. Blocking entity and routing to DLQ.%n", entityKey);
                        
                        // Mark this entity key as blocked so subsequent records for it don't jump the line
                        blockedEntities.add(entityKey);
                        
                        // Push the failed record to the DLQ
                        dlqProducer.send(new ProducerRecord<>("order-events-dlq", entityKey, record.value()));
                    }
                }
                consumer.commitSync();
            }
        }
    }
 
    private static void executeBusinessLogic(ConsumerRecord<String, String> record) throws Exception {
        if ("BAD_DATA".equals(record.value())) {
            throw new RuntimeException("Invalid processing payload");
        }
        System.out.println("Successfully processed: " + record.value());
    }
}

Would you like to explore how you eventually process and clear out messages from the order-events-dlq topic once the system is unblocked, or do you have questions about how the Redis/Database lookup would work at scale?

Backlinks