Some Random Notes

2. Ordering Coding

3 min read

Here are the boilerplate code snippets for both a Java producer and consumer, utilizing the official Apache Kafka clients.

1. The Idempotent Producer

This configuration guarantees that messages with the same key maintain their strict order within their designated partition, even during broker or network failures.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
 
import java.util.Properties;
 
public class OrderedProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        // ─── ORDERING & RELIABILITY CONFIGURATIONS ──────────────────────────────
        // 1. Ensures message sequencing and prevents retry-induced reordering
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        
        // 2. Requires all in-sync replicas to acknowledge the write
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        
        // 3. Retry indefinitely for transient errors (Idempotence makes this safe)
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        
        // 4. Maximum concurrent requests allowed while maintaining order (Max is 5 when Idempotence is true)
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
 
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            String topic = "order-events";
            
            // Critical: Use a non-null key (e.g., accountId or orderId) 
            // All events with key "customer_abc" are guaranteed to land on the same partition in exact sequence
            String key = "customer_abc"; 
            
            for (int i = 1; i <= 5; i++) {
                String value = "Event_" + i;
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
                
                producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        System.out.printf("Sent successfully to Partition: %d, Offset: %d%n", 
                                metadata.partition(), metadata.offset());
                    } else {
                        System.err.println("Failed to send record: " + exception.getMessage());
                    }
                });
            }
        } // Auto-closes and flushes remaining batches
    }
}

2. The Sequential Consumer

This consumer configuration processes messages sequentially on the single polling thread. It disables auto-commit, ensuring offsets are only committed after the business logic successfully completes.

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 OrderedConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
 
        // ─── ORDERING & OFFSET CONFIGURATIONS ───────────────────────────────────
        // Disable auto-commit to control exactly when an ordered message is marked "complete"
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 
        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) {
                    // Processing MUST stay single-threaded inside this loop.
                    // Do NOT pass this data to an asynchronous ThreadPool/ExecutorService,
                    // as threads execute at random speeds and will scramble event order.
                    processBusinessLogic(record);
                }
 
                // Commit synchronously after the entire batch of polled messages is successfully executed
                if (!records.isEmpty()) {
                    consumer.commitSync();
                }
            }
        }
    }
 
    private static void processBusinessLogic(ConsumerRecord<String, String> record) {
        System.out.printf("Processing Key: %s, Value: %s (Partition: %d, Offset: %d)%n",
                record.key(), record.value(), record.partition(), record.offset());
    }
}

Would you like me to translate these examples into a different programming language like Python (confluent-kafka) or Node.js, or are you interested in knowing how to handle partial ordering failure recovery (like using a dead-letter queue)?

Backlinks