Offsets and delivery guarantees

Kafka does not track which records a consumer has processed. It tracks which offset the consumer said it reached, and the consumer says so by committing. The gap between "processed" and "committed" is where every delivery guarantee lives: commit before you process and a crash loses records; commit after and a crash repeats them; make processing idempotent and repeats are harmless; use transactions and Kafka-to-Kafka pipelines become exactly-once. This lesson is offsets, the commit strategies, the message that was processed twice after a crash, and what exactly-once does and does not mean.

#Offsets

Each consumer group records, per partition, the offset of the next record to read, in an internal compacted topic called __consumer_offsets. When a consumer starts or takes over a partition, it fetches that offset and continues from it. When no offset exists (a new group, or one whose offset expired after offsets.retention.minutes, seven days by default), auto.offset.reset decides: latest (the default) skips everything already in the topic, earliest reads from the beginning. A new consumer group on a busy topic with the default silently starts from now, and the data before is never seen, which is a surprise on the first deploy of a new service and a design decision on the second.

The committed offset is not the position of the consumer's last read; the consumer keeps that in memory and reads ahead. It is a checkpoint: "everything before here is done". The commit strategy decides when that becomes true.

#Auto commit

enable.auto.commit=true (the default) commits the offsets returned by the previous poll() on the next poll(), if auto.commit.interval.ms (five seconds) has passed. It sounds like "commit every five seconds" and behaves like "commit whatever the last poll returned, whether or not you finished it":

1. poll() returns records 100–199.
2. The loop processes 100–149, and five seconds have passed.
3. The loop calls poll() again, which commits offset 200 first.
4. The process crashes while handling record 160.

Records 160–199 are committed and were never processed: lost, under a setting that looked safe. Auto commit is at-most-once with a delay. It is acceptable for metrics and logs where a gap is tolerable, and wrong for anything else.

#Manual commit and at-least-once

consumer.subscribe(List.of("orders"));
while (running) {
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, Order> r : records) {
        handle(r.value());                 // side effects happen here
    }
    consumer.commitSync();                 // only after every record in the batch is done
}
// crash between handle() and commitSync(): the batch is redelivered → at-least-once

With enable.auto.commit=false and a commit after processing, nothing is lost: a crash before the commit replays the uncommitted records on restart. The price is the mirror image: the message processed twice after a crash. The order was handled, the email sent, the row inserted, and then the process died before committing; on restart the same record arrives again and the handler runs again. At-least-once means exactly that, and the design question moves from "how do I commit" to "what happens when this runs twice".

commitSync() blocks until the coordinator acknowledges and retries on failure; commitAsync() returns immediately and does not retry (a later commit supersedes it anyway). The common pattern is async commits in the loop for throughput and one sync commit on shutdown and in the rebalance listener, so the last position is never lost. Committing per record is possible (commitSync(Map.of(partition, new OffsetAndMetadata(offset + 1)))) and expensive; committing per batch is the norm, and it makes the batch the unit of redelivery.

#The idempotent consumer

Since duplicates are guaranteed to occur, the consumer is written so that they do not matter:

@Transactional
public void handle(ConsumerRecord<String, Payment> r) {
    // the (topic, partition, offset) triple is unique forever; the event id works too
    int inserted = jdbc.update(
        "INSERT INTO processed (topic, partition, offset) VALUES (?, ?, ?) ON CONFLICT DO NOTHING",
        r.topic(), r.partition(), r.offset());
    if (inserted == 0) return;                    // seen it: the first run's effects are already committed
    accounts.credit(r.value().accountId(), r.value().amount());
}

The dedupe row and the business effect commit together, so either both happened or neither. A redelivered record finds its row and does nothing. This is the pattern for anything that writes to a database, and it is what people mean when they say "exactly-once" about a Kafka-to-database pipeline: at-least-once delivery plus an idempotent consumer. Where the effect is naturally idempotent (setting a status to SHIPPED, upserting a profile) the dedupe table is unnecessary; where it is not (crediting an account, sending an email) it is mandatory.

The strongest version stores the offset itself in the database with the effect, and on startup seeks the consumer to the stored offset instead of trusting __consumer_offsets. Then the database is the only source of truth for progress, and Kafka's commit is a hint.

#Exactly-once and transactions

Kafka's own exactly-once is narrower than the phrase suggests: it covers a consumer that reads from Kafka and a producer that writes to Kafka in the same application, so that the output records and the input offsets are committed atomically. A Kafka Streams job, or a consume-transform-produce service, can be configured so that a crash never produces duplicate output and never skips input:

producerProps.put("transactional.id", "enricher-" + instanceId);   // stable per instance
producer.initTransactions();
consumerProps.put("isolation.level", "read_committed");
 
while (running) {
    producer.beginTransaction();
    var records = consumer.poll(Duration.ofMillis(500));
    for (var r : records) producer.send(new ProducerRecord<>("enriched", r.key(), enrich(r.value())));
    producer.sendOffsetsToTransaction(currentOffsets(records), consumer.groupMetadata());
    producer.commitTransaction();        // output records and input offsets become visible together
}

Consumers of the output with isolation.level=read_committed never see records from an aborted transaction. What this does not cover is any side effect outside Kafka: a database write, an HTTP call, an email. Those still need the idempotent consumer. "Exactly-once" in Kafka means exactly-once between topics; end to end, it is at-least-once delivery with idempotent effects, and anyone who tells you otherwise has not had the incident.

#Choosing

#Under the hood: __consumer_offsets, the position, and the transaction coordinator

A commit is a produce. OffsetCommit sends, for each partition, a record keyed by (group, topic, partition) with the offset and metadata as its value to the internal, compacted topic __consumer_offsets, on the group's coordinator broker, with acks=all semantics. The coordinator also keeps the latest value per key in an in-memory cache, which is what an OffsetFetch reads when a consumer starts or takes over a partition. Compaction keeps the most recent commit per group-partition forever, until the group's offsets expire (offsets.retention.minutes, seven days after the group becomes empty, not seven days after the last commit) and a tombstone removes them. That is the mechanism behind "a group that was down for eight days starts from auto.offset.reset".

The consumer holds two numbers per partition, and they are not the same. The position is the next offset the consumer will fetch: it advances as poll() returns records, in memory, ahead of anything committed. The committed offset is whatever the last OffsetCommit said. commitSync() with no arguments commits the position, meaning "everything poll() has ever returned to me on this partition is done", which is exactly why committing before finishing the batch loses records and why the offset to commit for a single record is record.offset() + 1. Auto-commit is a check at the top of poll() that the interval has elapsed and, if so, an async commit of the positions; it runs before the new records are returned, so it is always the previous batch's high-water mark it commits.

Transactions add a second coordinator. initTransactions() registers the transactional.id with a transaction coordinator (the leader of a __transaction_state partition), which returns a PID and bumps the epoch, fencing any older producer with the same id. beginTransaction is local; the first send to each partition tells the coordinator to add it to the transaction. sendOffsetsToTransaction adds the group's __consumer_offsets partition as one more participant and writes the offsets there as an uncommitted record. commitTransaction is a two-phase protocol: the coordinator writes PREPARE_COMMIT to its own log, then writes a control record (a commit marker) into every participating partition, including the offsets partition, then COMPLETE_COMMIT. A read_committed consumer buffers records past the last stable offset (the first offset of any open transaction) until it sees the marker, and skips those under an abort marker; the offsets commit becomes visible with the same marker, which is what makes input progress and output atomic.

#Walkthrough: the credit applied twice

A ledger consumer credited accounts from a payments topic and, once a month, a customer was credited twice.

@KafkaListener(topics = "payments", groupId = "ledger")
public void on(Payment p, Acknowledgment ack) {
    accounts.credit(p.accountId(), p.amount());   // UPDATE accounts SET balance = balance + ? ; commits
    audit.record(p);                               // a second transaction
    ack.acknowledge();                             // MANUAL_IMMEDIATE commit
}

1. The handler was at-least-once by design: commit after the effect. The effect was two database transactions followed by an offset commit; a crash between the first and the third redelivered the record and ran credit again. Not idempotent: balance = balance + ? twice is twice the money.
2. The duplicates clustered on deploy days, when pods were killed mid-handler, and on one day when a rebalance revoked the partition after credit had committed but before acknowledge(): the new owner started from the last committed offset and replayed the record. No crash was needed, only a rebalance.
3. Idempotence via the producer was raised and dismissed correctly: the producer had sent each payment once, with one offset. The duplicate was the consumer's redelivery of a single record, which no producer setting addresses.
4. The fix put the dedupe key and the effect in one database transaction: INSERT INTO applied (payment_id) ... ON CONFLICT DO NOTHING, and only if that inserted a row, the credit and the audit row, all in the same @Transactional. A redelivered record found its applied row and returned without touching the balance. The offset commit stayed after the transaction.
5. The team also handled the rebalance case explicitly with a ConsumerAwareRebalanceListener that commits the current position on onPartitionsRevoked, so a clean revocation does not replay what was finished; the dedupe table remained the guarantee, since a crash skips the listener.

At-least-once means the handler will run twice for some record, and the only question is whether the second run changes anything. A dedupe key committed with the effect is how it does not.

#Try it yourself

#Misconceptions

• "Committed offset means last processed record." It is the next offset to read, and commitSync() with no arguments commits the position, which is ahead of processing.
• "Auto-commit commits every five seconds." It commits the previous poll's high-water mark at the start of the next poll, once the interval has elapsed. What that batch had finished is not considered.
• "Offsets expire seven days after the last commit." They expire seven days after the group becomes empty; an active group keeps them forever.
• "The idempotent producer prevents duplicate processing." It prevents duplicate records. Redelivery of one record to a consumer is unaffected and is the norm under at-least-once.
• "Kafka transactions make my database writes exactly-once." They make Kafka-to-Kafka atomic. A database write inside the loop is outside the transaction and needs its own dedupe.

#Going deeper

• Kafka documentation, "Consumer position", "Offset management" and the consumer configs enable.auto.commit, auto.offset.reset, isolation.level.
• KIP-98 and the design document "Exactly Once Delivery and Transactional Messaging in Kafka", for the transaction coordinator and control records.
• KIP-447 (producer scalability for exactly-once, sendOffsetsToTransaction with group metadata), which is the form shown in this lesson.
• Confluent, "Transactions in Apache Kafka" (Apurva Mehta, Jason Gustafson) and "Exactly-once semantics are possible: here's how".
• Kafka: The Definitive Guide, chapter 4 (consumers) and chapter 8 (exactly-once).