Offset Commits, and the Window That Decides Your Semantics

A Kafka consumer does not remember where it is. It tells the cluster where it is, by committing an offset — a number saying "the next record I have not dealt with is this one" — and if the process dies, that number is the only thing the replacement gets. Everything that matters about duplicates and lost records is the distance between that number and the work your code has actually finished.

So the question is never "auto-commit or manual commit". It is: at the instant the machine loses power, how far apart are those two things? The setting people reach for first — turning off enable.auto.commit and calling commitSync() after each batch — usually does not change that distance at all. With max.poll.records at its default of 500, a manual commit after every batch can leave a wider gap than the auto-commit it replaced.

Below is one partition, orders-3, and one consumer running a poll loop for twelve seconds. Three lines are drawn: the position (offsets poll() has handed to your code), the durable frontier (records whose side effect has actually landed), and the committed offset. Pick where the process dies with the crash at slider. Start with the defaults, then switch the commit strategy to commitSync() after the batch — and then push max.poll.records to 25 and watch the two strategies become identical.

At the defaults the consumer has been handed 45 records, finished 40 of them, and told the coordinator about 25. The crash at 8.2 s costs you fifteen records processed a second time, and the worst crash instant in those twelve seconds costs twenty-five. Switch to commitSync() after the batch and the worst case drops to five — one batch. That is the entire benefit of manual commit, and it is a real one.

Now set max.poll.records to 25. Both strategies report a worst case of twenty-five duplicates. Identical. Then drag auto.commit.interval.ms down to 100 and watch nothing happen. The duplicate window was never the interval; it was the batch.

What a commit actually is

A commit is a produce request. The consumer sends an OffsetCommit to its group coordinator — the broker that owns this group — and the coordinator appends a record to the internal __consumer_offsets topic, keyed by (group.id, topic, partition). There are 50 partitions in that topic by default and it is log-compacted, so the latest value for each key survives. That is the whole storage mechanism. Nothing about your consumer process is persisted; the only state that outlives it is one integer per partition per group.

That integer is the offset of the next record to read, not the last one read. The javadoc is explicit: "The committed offset should always be the offset of the next message that your application will read." So after processing offset 84,224 you commit 84,225. Off-by-one here is not a style question — commit the record you just finished and it is delivered again on every restart, forever.

Two positions exist and confusing them is the root of most of this. The position is in memory in your consumer and advances every time poll() hands you records. The committed offset is on a broker and advances only when you, or the client on your behalf, send an OffsetCommit. In the simulation those are the dashed line and the solid dark line. Nothing forces them to be close together.

Auto-commit does not run in the background

auto.commit.interval.ms defaults to 5000, and almost everyone reads that as "a background thread commits every five seconds". In the classic consumer — group.protocol=classic, still the default — there is no such thread. The check lives at the end of ConsumerCoordinator.poll():

maybeAutoCommitOffsetsAsync(timer.currentTimeMs());

public void maybeAutoCommitOffsetsAsync(long now) {
    if (autoCommitEnabled) {
        nextAutoCommitTimer.update(now);
        if (nextAutoCommitTimer.isExpired()) {
            nextAutoCommitTimer.reset(autoCommitIntervalMs);
            autoCommitOffsetsAsync();       // commits subscriptions.allConsumed()
        }
    }
}

Three consequences follow, and all three are visible in the simulation.

The interval is a floor, not a period. A commit can only happen at a poll() call. If your batch takes eight seconds to process, the 5000 ms timer expires four seconds into that batch and then waits four more seconds for the next poll(). Set max.poll.records to 25 and auto.commit.interval.ms to 100 in the simulation: the worst case stays at twenty-five duplicates, unchanged from 5000 ms. The commits are 5000 ms apart because the polls are.

The value committed is the position, not your progress. subscriptions.allConsumed() returns where poll() has read to. The consumer has no idea whether your handler succeeded, threw, or was never called. An exception you caught and logged does not hold the offset back.

In a plain synchronous loop this is still at-least-once. This is the part people get backwards. The commit fires at the top of a poll, committing positions from the previous poll — and by then the previous batch is finished, because the loop is what finished it. Leave the simulation on auto-commit with the worker pool unticked and drag the crash slider across all twelve seconds. The lost counter never leaves zero.

One version note. If you set group.protocol=consumer — the KIP-848 protocol, available since Kafka 4.0 — the client is the AsyncKafkaConsumer, and its auto-commit is driven by CommitRequestManager on the background network thread rather than by your poll call. The default is still classic, so unless you changed it, the behaviour above is yours.

Where auto-commit really does lose records

Tick hand each batch to a worker pool and poll again immediately, leave the strategy on auto-commit, and drag the crash to just after the commit mark at 5.0 s. The lost counter jumps to ten. The band under the lines turns from amber to magenta, because the committed offset is now above the durable frontier rather than below it. Ten records were handed to the pool, counted as position, committed — and never ran.

The javadoc says exactly why, in a sentence written about a different example: "If we allowed offsets to auto commit ... records would be considered consumed after they were returned to the user in poll." Returned to the user. Not processed by the user. As long as your loop is poll → process → poll, those two happen to coincide. The moment anything decouples them, they do not:

• an ExecutorService or any worker pool;
• a reactive pipeline where the handler returns a future;
• a batching writer that buffers records and flushes on a timer;
• an in-process queue between the consumer thread and the business logic;
• consumer.pause() plus a retry buffer.

Now switch the strategy to commitSync() after the batch, with the worker pool still ticked. Fifteen records lost — worse than auto-commit. This is the trap worth taking away from the whole lesson: "after the batch" means after the batch returned control to the loop. Once the loop's job is to enqueue, committing "after the batch" commits work that has not started. Manual commit did not make this safe; it made it at-most-once with a bigger window.

The librdkafka clients — and therefore the Python, Go, .NET and C++ ecosystems — split this into two settings for exactly this reason. enable.auto.offset.store controls whether reading a record marks it committable, and enable.auto.commit controls whether the stored value is sent. Setting the first to false and calling store_offsets() after processing gives you auto-commit's cheapness with manual commit's ordering. The Java client has no equivalent; there you commit manually or you accept the position.

The window is the batch

Set the worker pool back off. With a synchronous loop, the number of records at risk on a crash is:

window = max(records processed since the last commit)

and for each strategy that resolves to something you can read off your own configuration:

Read the middle row against the defaults: max.poll.records is 500. If a record takes 2 ms, a batch is one second of work, and manual per-batch commit gives you a one-second window against auto-commit's five. If a record takes 40 ms, a batch is twenty seconds of work, and manual per-batch commit is four times worse than the auto-commit you turned off — while also being the thing that makes you exceed max.poll.interval.ms and trigger a rebalance.

Per-record commit closes the window to one, and the price is visible. Set processing time per record to 50 ms and max.poll.records to 25, then compare the "loop time blocked on commits" readout between the two: 0.4% for per-batch, 9.1% for per-record. At a real 500-record batch that is 500 sequential round trips to one broker where there used to be one. The usual middle ground is to commit every n records or every k milliseconds, which is a window you chose rather than one the defaults chose for you.

commitAsync() is often proposed here as the fix for the cost. It is a latency fix, not a safety fix — the loop does not wait for the round trip, and if the response never comes the offset is simply not committed. The standard pattern is commitAsync() in the loop for speed and one commitSync() in a finally block on shutdown, so the last one is guaranteed. Note that commitAsync retries are dangerous: a retried commit that lands after a later commit would move the offset backwards, which is why the client does not retry them.

Exactly-once, and the three words that limit it

Switch the strategy to the transactional producer. Both counters go to zero at every crash instant, and the worst case reads 0 lost / 0 dup. That is real, and it is the only setting in the simulation that achieves it.

The construction is the one the Kafka design document describes: "The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the same transaction as the output topics." The producer holds a transactional.id, and each cycle is:

producer.beginTransaction();
for (ConsumerRecord<K,V> r : records) producer.send(transform(r));
producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();

If anything throws before commitTransaction(), the transaction aborts. The output records are still physically in the log, but they carry an abort marker, and a consumer with isolation.level=read_committed filters them out and refuses to read past the last stable offset — the offset of the first record belonging to a still-open transaction. The offsets written by sendOffsetsToTransaction are aborted with them, so the restart replays the batch. Two failure modes cancelling out.

Now the three words. It is exactly-once within Kafka.

• Your handler still runs twice. The simulation's "records re-read after restart" readout stays at five in transactional mode even though duplicates are zero. The replayed records go through your code again. Everything that code does outside the transaction — an HTTP POST, a row in Postgres, an email, an increment in Redis — happens a second time. The transaction can only roll back things Kafka wrote.
• The reader has to opt in. isolation.level defaults to read_uncommitted. A downstream consumer left at the default sees the aborted records as if they were fine. Every reader of every output topic in the chain has to be set to read_committed, and each one pays the latency of waiting for the last stable offset to advance.
• It is a throughput and latency trade. transaction.timeout.ms is 60 s by default and a hung transaction blocks read_committed readers of that partition until the coordinator aborts it. Small transactions mean frequent commit markers; large ones mean long blocking.

In Kafka Streams this is one line — processing.guarantee=exactly_once_v2, which is now the only exactly-once value the config accepts, requires brokers 2.5 or newer, and needs at least three brokers by default. The _v2 is KIP-447, which replaced "one producer per input partition" with fencing by consumer group metadata, and is why sendOffsetsToTransaction now takes a ConsumerGroupMetadata rather than a group id string.

Do not confuse any of this with enable.idempotence, which has defaulted to true since Kafka 3.0. Idempotence deduplicates a producer's own retries within one partition using a producer id and a sequence number. It stops a network timeout from writing the same record twice. It does nothing at all about consumer restarts.

Note also that read_committed here means something different from the isolation level of the same name in a database. There it decides which concurrent writers you can see; here it is a filter that hides records belonging to aborted or open transactions.

The commit that fails, and the rebalance

The simulation kills the process, which is the clean case: nothing else is contending for the partition. In production the commonest way to lose a commit is not a crash at all.

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be
completed since the group has already rebalanced and assigned the partitions
to another member. This means that the time between subsequent calls to poll()
was longer than the configured max.poll.interval.ms, which typically implies
that the poll loop is spending too much time message processing. You can
address this either by increasing max.poll.interval.ms or by reducing the
maximum size of batches returned in poll() with max.poll.records.

That text is the no-argument constructor of CommitFailedException, and it is accurate but incomplete about what it costs you. The sequence is: your batch took longer than max.poll.interval.ms (five minutes by default), the client proactively left the group, the coordinator gave orders-3 to another member, and your commitSync() was rejected because you no longer own the partition. Everything you processed in that batch is uncommitted, and the new owner is already replaying it. You get the duplicates without the crash.

This is why the fix for the exception is never "catch and retry the commit" — the partition is gone and retrying cannot succeed, which is why the class is documented as unrecoverable. The fix is to make the batch smaller or the interval longer, in that order.

Two smaller things in this family are worth knowing:

• Auto-commit runs during a rebalance too. onJoinPrepare issues one before revoking partitions, so a clean rebalance does not lose the interval's worth of work. Only an unclean exit does. If you commit manually, that safety net is yours to build with a ConsumerRebalanceListener and onPartitionsRevoked. Getting the group to stop rebalancing at all on a rolling restart is what <code>group.instance.id</code> is for.
• Committed offsets expire. offsets.retention.minutes defaults to 10080 — seven days — measured from when the group became empty. Stop a consumer for a long weekend plus a bit and the offsets are gone; on restart auto.offset.reset decides what happens, and its default is latest, which silently skips everything produced while you were down. This is the one way to lose records without a single commit bug.

Reading it on a real cluster

The gap in the simulation has a direct equivalent you can measure. Run:

kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group order-processor --describe

TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
orders   3          84225           84512           287  consumer-1-a3f...

CURRENT-OFFSET is the committed offset — the dark line. It is not where the consumer is; it is where the consumer would restart. To see the other line you need the client, because the position never leaves the process. The JMX metric is kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,topic=orders,partition=3 and the attribute is records-lag, computed from the position. If records-lag is materially smaller than the LAG column from the CLI, the difference is your uncommitted window, in records. That single subtraction is the diagnosis, and it is the reason lag from the CLI and lag from the client are different numbers.

To find out whether you are committing at all, and how often:

kafka-consumer-groups.sh --describe --group order-processor --offsets
# run it twice, ten seconds apart, and diff CURRENT-OFFSET

# or read the offsets topic directly
kafka-console-consumer.sh --bootstrap-server broker:9092 \
  --topic __consumer_offsets --from-beginning \
  --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter"

The second command prints one line per commit with its timestamp. If you see a commit every five seconds you are on auto-commit whether you configured it or not. If you see one per batch, someone is calling commitSync(). This is the fastest way to answer "which layer of this framework owns my commits", which is a question almost nobody using Spring Kafka, Flink or Kafka Connect can answer from their own configuration.

On the client side, three metrics from consumer-coordinator-metrics tell the rest: commit-latency-avg and commit-latency-max (a commitSync in the loop puts this directly into your throughput), commit-rate (compare it to your poll rate — equal means per-batch, much lower means auto-commit), and failed-rebalance-rate-per-hour, because every rebalance is a replay of every group member's uncommitted window at once.

Finally, the decision that actually removes the problem. Duplicates are cheap to design for and expensive to eliminate: give each record a stable key, make the handler idempotent — INSERT ... ON CONFLICT DO NOTHING, an upsert keyed on (topic, partition, offset), a conditional write — and then the size of the window stops mattering. Loss is the opposite: there is no downstream repair for a record nobody processed, because nothing knows it existed. So the ordering rule is worth stating on its own line: make the effect durable, then commit. Every configuration in the simulation that obeys it is at-least-once, and every one that violates it is at-most-once.

Next, the two mechanisms that decide when a replacement consumer shows up to replay your uncommitted window at all: what a rebalance stops and for how long and how partitions get handed between members. And if the duplicates only ever appear during a deploy, static membership removes the rebalance that causes them.