03 - Advanced Kafka Producer and Consumer: Acks, Idempotency and Retry

Does an order arrive once, twice or never? It depends on six properties

Your checkout service sends an order to Kafka. The broker writes it, the reply gets lost on the way, the client times out and retries. A possible result: the topic holds two identical orders and the customer is charged twice. Or the opposite: the leader acknowledges and dies before a follower copies the record, and the order vanishes. These are two different failures and they are avoided by two different mechanisms, acks with min.insync.replicas for durability and idempotence for duplicates. On the reading side the risk is symmetrical: an offset saved too early makes records get skipped, one saved too late makes them be processed again.

The useful news is that the Kafka 4.4 documentation (consulted on 2026-11-12) already sets the safe combination by default. The listing below is real code: it reads the producer configuration of the Java client 4.4.0 (kafka-clients-4.4.0.jar from Maven Central, SHA-1 verified) without connecting to any broker, and prints the values you would have if you wrote nothing.

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.ConfigException;
import java.util.*;

public class ConfigProducer {
    static Map<String, Object> base() {
        Map<String, Object> p = new HashMap<>();
        p.put("bootstrap.servers", "localhost:9092");   // not contacted: only the configuration is read
        p.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return p;
    }

    public static void main(String[] args) {
        ProducerConfig defaults = new ProducerConfig(base());
        for (String k : List.of("acks", "enable.idempotence", "retries", "delivery.timeout.ms",
                "request.timeout.ms", "linger.ms", "batch.size", "max.in.flight.requests.per.connection"))
            System.out.println(k + " = " + defaults.values().get(k));

        Map<String, Object> conflict = base();
        conflict.put("enable.idempotence", true);      // explicit request...
        conflict.put("acks", "1");                     // ...incompatible with acks=1
        try {
            new ProducerConfig(conflict);
        } catch (ConfigException e) {
            System.out.println("ConfigException: " + e.getMessage());
        }
    }
}

// Output:
// acks = -1
// enable.idempotence = true
// retries = 2147483647
// delivery.timeout.ms = 120000
// request.timeout.ms = 30000
// linger.ms = 5
// batch.size = 16384
// max.in.flight.requests.per.connection = 5
// ConfigException: Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.

The first eight lines are the defaults. acks appears as -1, which is the numeric form of all; enable.idempotence is true; retries is 2147483647 (practically "retry until time runs out"), because the real limit is delivery.timeout.ms, 120000 ms; max.in.flight.requests.per.connection is 5, the maximum compatible with idempotence. linger.ms at 5 is a novelty: the upgrade notes say the default changed from 0 to 5 in Kafka 4.0, because larger batches generally do not worsen latency. The last line is the instructive part: if you explicitly ask for idempotence and set acks=1, the client rejects the configuration with a ConfigException. If instead you do not mention idempotence and set acks=1, the documentation says idempotence is simply disabled instead of failing: it is the most common way of losing a guarantee while believing you have it.

The three delivery guarantees, and what each one costs

The "Message Delivery Semantics" section of the documentation distinguishes at most once (a message may be lost but is not duplicated), at least once (it is not lost but may be duplicated) and exactly once (one single time). Kafka's default is at-least-once; to obtain at-most-once you must disable the producer's retries and save the offset before processing the batch. Exactly-once, for a flow that reads from Kafka and writes to Kafka, requires transactions, which the next article covers. Towards an external system (a database, an API) the documentation is clear: that system's cooperation is needed, for example a deduplication key.

The choice depends on the damage an error causes. A click counter can lose a few events (at-most-once is tolerable); a charge can neither be lost nor duplicated (you need deduplication, and the producer's idempotence is only the first link). Remember that the producer's idempotence protects against duplicates in the log; it says nothing about what the consumer does after reading.

acks and min.insync.replicas: when a record is safe

acks establishes how many acknowledgements the producer waits for. With acks=0 it waits for nothing: the record is considered sent when it enters the socket buffer, the returned offset is always -1 and retries have no effect, because the client does not know whether anything went wrong. With acks=1 it waits for the leader, which can acknowledge and fail before the followers copy. With acks=all it waits for all the replicas of the ISR, but "all" can mean just one, the leader, if the ISR has shrunk. That is why the second parameter is needed, min.insync.replicas (default 1 in the topic documentation): with acks=all, if the ISR has fewer members than that minimum, the producer receives an error (NotEnoughReplicas or NotEnoughReplicasAfterAppend). With replication factor 3 and a minimum of 2, an acknowledged record exists on at least two brokers (the ISR is explained in the article on the fundamentals).

The price is availability: if two brokers out of three fail, writes stop instead of accepting a record with a single copy. It is a choice, not a defect, and it must be made consciously for topics that contain data you cannot regenerate. How much extra latency it costs has no absolute valid figure: it depends on the network and disks of your cluster, and must be measured with a test of your own.

Retries and duplicates: the model

The case that turns retries into duplicates is the most insidious: the broker writes and acknowledges, but the acknowledgement does not get back to the client. From the producer's point of view there is a record with no reply and it does not know whether it was written. The following model compares three producers facing two failures: the request that gets lost before arriving and the acknowledgement that gets lost on the way back. The model's broker writes and, if the producer is idempotent, remembers the last sequence seen for each producer id and ignores those already seen.

import java.util.*;

public class Delivery {
    enum Failure { REQUEST_LOST, ACK_LOST }

    static class Broker {
        final List<String> log = new ArrayList<>();
        final Map<Long, Integer> lastSequence = new HashMap<>();   // for each producer id

        // Returns true if the broker acknowledges. With idempotence it discards sequences already seen.
        boolean write(long pid, int seq, String value, boolean idempotent) {
            if (idempotent && seq <= lastSequence.getOrDefault(pid, -1)) return true;  // duplicate
            log.add(value);
            lastSequence.put(pid, seq);
            return true;
        }
    }

    // Sends a record; on the first attempt the indicated failure happens. The second attempt succeeds.
    static List<String> attempt(Failure failure, boolean retries, boolean idempotent) {
        Broker broker = new Broker();
        long pid = 7;
        int seq = 0;                                           // same sequence on every attempt
        if (failure == Failure.REQUEST_LOST) {
            // the request never reaches the broker: no write
        } else {
            broker.write(pid, seq, "R1", idempotent);         // the broker writes, but the ack is lost
        }
        if (retries) broker.write(pid, seq, "R1", idempotent);   // the producer got no reply
        return broker.log;
    }

    public static void main(String[] args) {
        String[] names = {"no retry            ", "retry, no idempot.  ", "retry + idempotence "};
        boolean[][] mode = {{false, false}, {true, false}, {true, true}};
        System.out.println("                     request lost | ack lost");
        for (int i = 0; i < 3; i++) {
            System.out.println(names[i] + " " + attempt(Failure.REQUEST_LOST, mode[i][0], mode[i][1])
                + " | " + attempt(Failure.ACK_LOST, mode[i][0], mode[i][1]));
        }
    }
}

// Output:
//                      request lost | ack lost
// no retry             [] | [R1]
// retry, no idempot.   [R1] | [R1, R1]
// retry + idempotence  [R1] | [R1]

The table says it all in six cells. Without retries, a lost request means a lost record ([]) while a lost acknowledgement leaves a record written, because the broker already had it. With retries, the lost request is resolved but the lost acknowledgement produces [R1, R1]: the duplicate. With idempotence both columns give [R1], because the second attempt carries the same pair (producer id, sequence) and the broker recognizes it. A real broker does much more (it keeps the last five requests per producer, rejects out-of-order sequences), but the principle is this: deduplication lives on the broker, not in your code.

There is a limit to know. The producer's identity holds for the session: if the process restarts it obtains a new producer id and the broker cannot recognize as a duplicate what was sent before the restart. The KafkaProducer Javadoc says it this way: "the producer can only guarantee idempotence for messages sent within a single session". Only with a transactional.id do you get a recovery that crosses several sessions, and it is the subject of the article on transactions. The Javadoc again confirms that since 3.0 enable.idempotence is true by default.

Order and in-flight requests

To go fast the producer keeps several requests in flight towards the same broker (max.in.flight.requests.per.connection, 5). If the first fails and the second does not, the first one's retry arrives afterwards: the two requests are swapped. The documentation of the retries parameter says so explicitly: with retries on, idempotence off and more than one request in flight, "the records in the second batch may appear first".

import java.util.*;

public class Ordering {
    static class Broker {
        final List<String> log = new ArrayList<>();
        final boolean idempotent;
        int expected = 0;                                  // next sequence accepted
        Broker(boolean idempotent) { this.idempotent = idempotent; }

        boolean receive(int seq, String value) {
            if (idempotent) {
                if (seq != expected) return false;        // out of sequence: rejected
                expected++;
            }
            log.add(value);
            return true;
        }
    }

    // Two requests in flight to the same partition: B1 (sequence 0) is lost, B2 (sequence 1) arrives.
    static List<String> simulate(boolean idempotent) {
        Broker broker = new Broker(idempotent);
        boolean b2Accepted = broker.receive(1, "B2");     // arrives first
        broker.receive(0, "B1");                          // the producer retries B1
        if (!b2Accepted) broker.receive(1, "B2");         // rejected: it resends it, now in the right order
        return broker.log;
    }

    public static void main(String[] args) {
        System.out.println("retry, no idempotence:   " + simulate(false));
        System.out.println("retry, with idempotence: " + simulate(true));
    }
}

// Output:
// retry, no idempotence:   [B2, B1]
// retry, with idempotence: [B1, B2]

Without idempotence the log ends up in the order [B2, B1]. With idempotence the model's broker accepts only the expected sequence: it rejects B2 (sequence 1 while it expects 0), then accepts B1 and finally the retried B2, so [B1, B2]. Here too lies the reason for the cap at 5: the documentation explains that the broker keeps at most five batches per producer, and if the in-flight requests were more, "previous batches may be removed on broker side". Since 4.0, as the upgrade notes say, the idempotence setting no longer falls back by itself when max.in.flight.requests.per.connection exceeds 5.

Timeout: the ceiling that governs retries

The setting that decides how long to keep retrying is delivery.timeout.ms (120000 ms), not retries: it is the upper bound of time between the return of send() and the outcome, retries included. The documentation of retries recommends leaving it at the default value and governing the behaviour with the timeout, which must be at least the sum of request.timeout.ms (30000 ms) and linger.ms. Choose the timeout based on how long you can wait before telling the user it failed: 120 seconds is an eternity for a web request, and for a checkout it is probably better to lower it and handle the error in the callback.

The consumer: where the position is saved

With enable.auto.commit=true (the default, every 5000 ms) the offset is saved inside the call to poll(), not by a timer halfway through processing. The KafkaConsumer Javadoc specifies the condition for staying at-least-once: you must consume all the records returned by each poll() before the next one. If you do not, the Javadoc writes, the saved offset can exceed the position actually consumed, with skipped records. The typical case is processing handed off to another thread. The model shows the two cases.

import java.util.*;

public class AutoCommit {
    // Consumer model: auto-commit saves the position inside poll(), not in the background.
    static class Consumer {
        long position = 0, committed = 0;
        List<Long> poll(int count) {
            committed = position;                            // auto-commit: saves what was returned BEFORE
            List<Long> batch = new ArrayList<>();
            for (int i = 0; i < count; i++) batch.add(position++);
            return batch;
        }
    }

    public static void main(String[] args) {
        // Case 1: the whole batch is processed before the next poll, then the process dies halfway through the second.
        Consumer c1 = new Consumer();
        List<Long> first = c1.poll(5);                       // offsets 0..4
        List<Long> processed = new ArrayList<>(first);       // all five processed
        List<Long> second = c1.poll(5);                     // the commit of 5 happens here
        processed.addAll(second.subList(0, 2));             // 5 and 6 processed, then crash
        System.out.println("case 1: committed=" + c1.committed + ", processed=" + processed
            + " -> restarts from " + c1.committed + ", reprocesses 5 and 6, nothing lost");

        // Case 2: the batch goes to another thread, the next poll starts immediately.
        Consumer c2 = new Consumer();
        List<Long> batch = c2.poll(5);                      // 0..4 handed to the worker
        List<Long> done = new ArrayList<>(batch.subList(0, 3));   // the worker has finished 0, 1, 2
        c2.poll(5);                                         // commit of 5 while the worker is halfway
        List<Long> lost = new ArrayList<>(batch.subList(3, 5));
        System.out.println("case 2: committed=" + c2.committed + ", processed=" + done
            + " -> restarts from " + c2.committed + ", lost offsets: " + lost);
    }
}

// Output:
// case 1: committed=5, processed=[0, 1, 2, 3, 4, 5, 6] -> restarts from 5, reprocesses 5 and 6, nothing lost
// case 2: committed=5, processed=[0, 1, 2] -> restarts from 5, lost offsets: [3, 4]

In the first case the commit of 5 happens at the start of the second poll, when the whole first batch has already been processed; the crash after offset 6 forces reprocessing 5 and 6, and nothing is lost. In the second the worker has finished only 0, 1 and 2 when the next poll saves 5: after the crash you restart from 5 and offsets 3 and 4 will never be processed. The model is not the real consumer (it has neither the 5000 ms timer nor the fetch), but it follows the Javadoc's rule, and the lesson is that auto-commit is safe only if processing is sequential. Otherwise switch to manual commit.

Manual commit has two forms: commitSync() saves the current position after processing, commitSync(Map) saves explicit offsets per partition. In both the right value is record offset + 1. The listing below uses the MockConsumer of the Kafka client 4.4.0, which runs the library's real code without a broker: it processes a batch, saves with commitSync, processes a second batch without saving it and simulates a rebalance that removes a partition. The ConsumerRebalanceListener saves the offsets of the partition it is about to lose.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;

public class ManualCommit {
    public static void main(String[] args) {
        MockConsumer<String, String> consumer = new MockConsumer<>("earliest");   // real client, no broker
        TopicPartition p0 = new TopicPartition("orders", 0), p1 = new TopicPartition("orders", 1);
        Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();

        consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
            @Override public void onPartitionsRevoked(Collection<TopicPartition> lost) {
                Map<TopicPartition, OffsetAndMetadata> mine = new HashMap<>();
                for (TopicPartition tp : lost)
                    if (toCommit.containsKey(tp)) mine.put(tp, toCommit.remove(tp));
                if (!mine.isEmpty()) consumer.commitSync(mine);        // save before losing the partition
                System.out.println("revoked " + lost + ", saved " + offsets(mine));
            }
            @Override public void onPartitionsAssigned(Collection<TopicPartition> added) {
                System.out.println("assigned " + added);
            }
        });
        consumer.rebalance(List.of(p0, p1));
        consumer.updateBeginningOffsets(Map.of(p0, 0L, p1, 0L));
        addRecords(consumer, 0, 3);
        process(consumer, toCommit);
        consumer.commitSync(toCommit);                              // manual commit, after processing
        System.out.println("committed: " + offsets(consumer.committed(Set.of(p0, p1))));
        toCommit.clear();

        addRecords(consumer, 3, 5);
        process(consumer, toCommit);                                // processed but NOT yet saved
        System.out.println("to save: " + offsets(toCommit));
        consumer.rebalance(List.of(p0));                               // p1 moves to another consumer
    }

    static void addRecords(MockConsumer<String, String> c, long from, long to) {
        for (long i = from; i < to; i++) {
            c.addRecord(new ConsumerRecord<>("orders", 0, i, "a", "order-0-" + i));
            c.addRecord(new ConsumerRecord<>("orders", 1, i, "b", "order-1-" + i));
        }
    }

    static void process(MockConsumer<String, String> c, Map<TopicPartition, OffsetAndMetadata> toCommit) {
        for (ConsumerRecord<String, String> r : c.poll(Duration.ofMillis(10))) {
            // the business logic would go here; then note the NEXT offset to read
            toCommit.put(new TopicPartition(r.topic(), r.partition()), new OffsetAndMetadata(r.offset() + 1));
        }
    }

    static String offsets(Map<TopicPartition, OffsetAndMetadata> m) {
        TreeMap<String, String> sorted = new TreeMap<>();
        m.forEach((tp, o) -> sorted.put(tp.toString(), o == null ? "null" : String.valueOf(o.offset())));
        return sorted.toString();
    }
}

// Output:
// assigned [orders-0, orders-1]
// committed: {orders-0=3, orders-1=3}
// to save: {orders-0=5, orders-1=5}
// revoked [orders-1], saved {orders-1=5}
// assigned []

The second line shows the committed offset after the first batch: 3 for each partition, that is, the offset of the last record (2) plus one. After the second batch the offsets to save are 5, but nobody has committed them yet. The simulated rebalance revokes orders-1 and the listener saves it: "saved {orders-1=5}". Whoever receives that partition will restart from 5 and will not reprocess records 3 and 4. The MockConsumer is a test double, so it does not demonstrate how the coordinator behaves on a real cluster; it demonstrates the correct use of the API and its contract.

Rebalance: stop everyone or stop only whoever changes

When a consumer joins or leaves, the group reassigns the partitions. With the classic protocol the behaviour depends on the assignor. The default list of partition.assignment.strategy is [RangeAssignor, CooperativeStickyAssignor] and, as the documentation says, the group uses the RangeAssignor until you do a rolling restart switching to the cooperative one: the RangeAssignor performs an eager rebalance, in which every consumer releases all its partitions and the group stops. The CooperativeStickyAssignor (KIP-429) revokes only the partitions that change owner.

import java.util.*;

public class Rebalance {
    static final int PARTITIONS = 6;

    public static void main(String[] args) {
        // State before: c1 has 0,1,2 and c2 has 3,4,5. c3 joins: the target is 2 partitions each.
        Map<String, List<Integer>> before = new LinkedHashMap<>();
        before.put("c1", List.of(0, 1, 2));
        before.put("c2", List.of(3, 4, 5));
        int target = PARTITIONS / 3;

        // Eager: everyone releases everything, then it is reassigned from scratch.
        int revokedEager = PARTITIONS;

        // Cooperative (sticky model): everyone keeps up to 'target' partitions and gives the rest to c3.
        List<Integer> given = new ArrayList<>();
        for (List<Integer> mine : before.values())
            given.addAll(mine.subList(target, mine.size()));

        System.out.println("eager:       revoked " + revokedEager + " partitions out of " + PARTITIONS
            + ", the whole group stops");
        System.out.println("cooperative: revoked " + given.size() + " partitions " + given
            + ", the other " + (PARTITIONS - given.size()) + " keep reading");
    }
}

// Output:
// eager:       revoked 6 partitions out of 6, the whole group stops
// cooperative: revoked 2 partitions [2, 5], the other 4 keep reading

With six partitions and a third member that must receive two, the eager rebalance stops the whole group and revokes six partitions; the cooperative one revokes only two (2 and 5), while the other four keep reading. The model uses a simplified sticky criterion and ignores the second round of messages of the real protocol, which serves to reassign the revoked partitions. There is also the new protocol (KIP-848), activated with group.protocol=consumer: the assignment is decided by the coordinator on the broker and there is no longer a global barrier. The default remains classic in the 4.4 documentation, so you have to choose it yourself.

One setting remains that causes "inexplicable" rebalances: max.poll.interval.ms (300000 ms). If your loop does not call poll() within that interval, the consumer leaves the group on its own initiative and the next commitSync fails with CommitFailedException. The Javadoc proposes two levers: lower max.poll.records (default 500) to shorten the work per iteration, or move the processing to another thread, remembering however the auto-commit rule seen above.

What to really tune

The table collects the tuning parameters that make sense to touch, with the defaults of the 4.4 documentation. No throughput figure here: every value depends on the load, and the only honest way to choose it is to test.

A rule from the producer documentation is worth more than any table: compression acts on whole batches, so it depends on linger.ms and batch.size. Choose an algorithm (gzip, snappy, lz4, zstd) only after measuring CPU and message size on your data.

What you can do now

You can read the default configuration of a 4.4 producer and know why it is already safe, and you can recognize the three ways of breaking it: acks=1 with idempotence not declared, more than five in-flight requests, min.insync.replicas left at 1. You can explain a duplicate as a lost acknowledgement and a reordering as an out-of-sequence retry. On the consumer you know when auto-commit is safe, you write a manual commit with the offset +1 and a listener that saves before losing a partition, and you know that the default classic rebalance stops the whole group. For atomic writes across several topics move on to transactions; for messages you cannot process, to the dead letter queue. If you arrive from another point of the series, the links are: KRaft and the controllers, the Schema Registry, monitoring, Kafka in production.