Some Random Notes

5. Drain and Catch Up

3 min read

To resolve and clear out messages from the DLQ (Dead Letter Queue) while restoring order, you must follow a strict process: you cannot simply process the DLQ in the background while your main consumer is still running. If you do, new messages and old failed messages will conflict with each other.

To unblock the queue safely, architectures use a Drain and Catch Up process.

The Unblocking Process (Step-by-Step)

  1. Fix the Root Cause: A developer updates the consumer code to handle the bad payload, or a DBA restores a failing database.
  2. Stop the Main App: Pause or shut down the main consumer group so it stops processing new incoming records for the blocked entity.
  3. Process the DLQ: Run a specialized “DLQ Recovery Worker” that reads the failed messages sequentially from the DLQ topic and executes them.
  4. Resume Normal Operations: Once the DLQ is completely empty for that entity, clear the Blocked Flag in your tracking database (or Redis) and turn the main consumer back on.

The Code: DLQ Recovery Worker

Here is how a recovery system reads messages from the order-events-dlq topic, processes them, and unblocks the key once the backlog is clean.

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;
 
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
 
public class DLQRecoveryWorker {
 
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "dlq-recovery-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 
        try (KafkaConsumer<String, String> dlqConsumer = new KafkaConsumer<>(props)) {
            // 1. Subscribe to the dead-letter topic
            dlqConsumer.subscribe(Collections.singletonList("order-events-dlq"));
            System.out.println("DLQ Recovery Worker started. Standing by for records...");
 
            while (true) {
                ConsumerRecords<String, String> records = dlqConsumer.poll(Duration.ofMillis(500));
 
                if (records.isEmpty()) {
                    // In a production app, if the DLQ stays empty for a long time, 
                    // it indicates that recovery might be finished for the active batch.
                    continue;
                }
 
                for (ConsumerRecord<String, String> record : records) {
                    String entityKey = record.key();
                    
                    try {
                        // 2. Retry the business logic (Root cause must be fixed by now!)
                        retryBusinessLogic(record);
                        
                        // 3. Unblock the entity after its backlog is cleared
                        // This allows the main application consumer to process its new messages normally again
                        removeFromBlockedListInDatabase(entityKey);
                        
                        System.out.printf("Successfully recovered message for key: %s. Entity unblocked.%n", entityKey);
                    } catch (Exception e) {
                        // If it fails AGAIN, stop immediately. Do not commit.
                        System.err.printf("CRITICAL: Recovery failed for key %s. Reason: %s. Halting worker.%n", 
                                entityKey, e.getMessage());
                        return; 
                    }
                }
                
                // Only commit DLQ offsets if the entire batch was processed flawlessly
                dlqConsumer.commitSync();
            }
        }
    }
 
    private static void retryBusinessLogic(ConsumerRecord<String, String> record) {
        // Now running with the bugfix or recovered database connection
        System.out.printf("Re-processing DLQ message: Key=%s, Value=%s%n", record.key(), record.value());
    }
 
    private static void removeFromBlockedListInDatabase(String key) {
        // Code to delete the block-flag from your shared store (Redis/Database)
        System.out.printf("Removed external database block for key: %s%n", key);
    }
}

Would you like to explore how to automate this recovery using Spring Kafka’s Container pausing features, or do you want to move on to how Kafka transaction settings (isolation.level) play into strict message ordering?

Backlinks