Producers & Consumers

A complete producer#

This publishes an orders event, keyed by orderId so all of one order's events land in the same partition (see Topics, Partitions & Offsets for why that matters):

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class OrderEventProducer {
    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());

        // Durability-related configs — see below for what each one actually buys you
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            String orderId = "order-482";
            String payload = "{\"orderId\":\"" + orderId + "\",\"status\":\"CREATED\"}";

            ProducerRecord<String, String> record = new ProducerRecord<>("orders", orderId, payload);

            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Failed to send record: " + exception.getMessage());
                } else {
                    System.out.printf("Sent to partition %d at offset %d%n",
                        metadata.partition(), metadata.offset());
                }
            });
        }
    }
}

producer.send() is asynchronous — it returns immediately and hands the record to a background batching thread. The callback (or the Future<RecordMetadata> it also returns) is how you find out whether the send actually succeeded, and on which partition/offset it landed.

The acks setting, concretely#

acks controls how many broker replicas must confirm a write before the producer considers it successful — it's the single biggest lever on the durability-vs-latency trade-off:

ENABLE_IDEMPOTENCE_CONFIG = true deduplicates retries at the broker level — if a retry caused by a network blip actually did succeed the first time, the broker recognizes the duplicate attempt (via a producer ID + sequence number) and doesn't write it twice. Combined with acks=all, this gets you Kafka's strongest per-partition delivery guarantee without writing any dedup logic yourself.

A complete consumer#

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class OrderEventConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "inventory-service");
        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); // commit manually — see below
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

                for (ConsumerRecord<String, String> record : records) {
                    reserveStockFor(record.key(), record.value());
                }

                consumer.commitSync(); // only commit after the whole batch is actually processed
            }
        }
    }

    private static void reserveStockFor(String orderId, String eventJson) {
        // business logic here
    }
}

poll() is the heart of the consumer loop — it both fetches new records and signals to the broker that this consumer is still alive (a consumer that never calls poll() within max.poll.interval.ms gets treated as dead and evicted from the group, triggering a rebalance).

Manual vs. auto commit#

With ENABLE_AUTO_COMMIT_CONFIG = true (the default), the client commits the latest offset on a timer, regardless of whether your code actually finished processing those records. That creates two failure windows:

• Crash after the auto-commit timer fires, before processing finishes → those records are marked as done but never actually got processed — silent data loss.
• Crash after processing finishes, before the next auto-commit fires → those records get reprocessed after restart — a duplicate.

Manual commit (commitSync() or commitAsync()) lets you commit only once you know processing actually succeeded, trading a small amount of code for control over exactly when "done" is recorded. commitSync() blocks until the broker confirms and retries on retriable errors — simplest and safest. commitAsync() doesn't block the poll loop but can complete out of order if a later commit's response comes back before an earlier one's, so most real consumers use commitAsync() on every iteration for throughput and a final commitSync() in a shutdown hook to guarantee the last batch is durably committed.

Either way, at-least-once delivery is still the practical default — a crash between "processed" and "committed" always reprocesses that batch on restart, so consumers should be written to be idempotent regardless of which commit strategy you pick.

Common mistakes#

• Leaving enable.auto.commit=true for work where losing or duplicating a record actually matters, without understanding which of the two failure windows above you're exposed to.
• Doing slow, blocking work inside the poll loop (a synchronous call to another service, for example) without accounting for max.poll.interval.ms — exceed it, and the consumer gets kicked from the group mid-processing, triggering a rebalance and, depending on commit timing, reprocessing of what it was just working on.
• Assuming acks=all alone means "no duplicates" — it protects against lost writes on the producer side, not consumer-side duplicate processing; that still needs an idempotent consumer or enable.idempotence for the producer's own retries.
• Catching and swallowing exceptions inside the record-processing loop without any dead-letter or retry strategy — a single poison record can silently be dropped forever, or block the entire partition forever, depending on how the exception is handled.