Some Random Notes

3. Partial ordering failure recovery

3 min read

Here is how you handle partial ordering failure recovery when keeping strict sequential order is critical.

When a message processing fails in a consumer, you cannot simply skip it or send it straight to a standard Dead-Letter Queue (DLQ). If you process subsequent messages for that same key while the failed message waits in a DLQ, you have broken message ordering.

To handle errors safely without permanently blocking the entire partition, architectures use a pattern called Ordered DLQ / Letterbox Routing or Redirected Processing Loops.

The Failure Strategy

  1. Detect Failure: A message fails business processing.
  2. Publish to DLQ with Expiry: Route the failed message to a DLQ, but pause consumption for that specific partition or route subsequent messages for that specific key to a hold state. Alternatively, route all subsequent messages for that specific entity key to the DLQ behind the failed message to preserve their relative order.
  3. Stop & Retry (Circuit Breaker): For global partition ordering, pause the consumer, back off, and retry processing the exact same offset.

Here is a Java implementation using a Retry-with-Backoff Circuit Breaker approach. If a record fails, it loops and retries up to a maximum limit before safely halting or logging, without advancing the offset.

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.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
 
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
 
public class ResilientOrderedConsumer {
    private static final int MAX_RETRY_ATTEMPTS = 3;
 
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "resilient-order-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");
 
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("order-events"));
 
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
 
                for (ConsumerRecord<String, String> record : records) {
                    boolean success = processWithRetry(record);
 
                    if (!success) {
                        // ─── CRITICAL ORDERING BREAKDOWN ─────────────────────────────────
                        // Message failed all retries. To guarantee strict order, we CANNOT 
                        // move to the next message. We must stop processing this partition.
                        System.err.printf("CRITICAL: Order blocked at partition %d, offset %d. Routing to alert system.%n", 
                                record.partition(), record.offset());
                        
                        // Seek back to this specific failed offset so it retries on the next poll cycle
                        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                        consumer.seek(tp, record.offset());
                        
                        // Break out of processing the current batch to prevent executing next records
                        break; 
                    }
 
                    // Explicitly commit individual offsets only after successful execution
                    TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                    OffsetAndMetadata offsetMeta = new OffsetAndMetadata(record.offset() + 1);
                    consumer.commitSync(Collections.singletonMap(tp, offsetMeta));
                }
            }
        }
    }
 
    private static boolean processWithRetry(ConsumerRecord<String, String> record) {
        int attempts = 0;
        while (attempts < MAX_RETRY_ATTEMPTS) {
            try {
                attempts++;
                // Simulate business logic
                executeBusinessLogic(record);
                return true; // Success!
            } catch (Exception e) {
                System.err.printf("Attempt %d failed for key %s: %s. Retrying...%n", 
                        attempts, record.key(), e.getMessage());
                
                // Exponential backoff before retrying the exact same message
                try {
                    Thread.sleep(1000L * attempts); 
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    return false;
                }
            }
        }
        return false; // All retry attempts exhausted
    }
 
    private static void executeBusinessLogic(ConsumerRecord<String, String> record) throws Exception {
        // Example failure simulator
        if ("Event_3".equals(record.value())) {
            throw new RuntimeException("Database timeout error");
        }
        System.out.printf("Successfully processed: %s%n", record.value());
    }
}

Would you like to explore how to implement an Entity-based Hold State (where you block only the failed customer_id but let other customers’ data process normally), or do you want to look at how Spring Kafka handles this natively using its DefaultErrorHandler?

Backlinks