Committing an Offset

Concepts covered: Offset commits

The committed offset is the next offset to read

A consumer's position lives in its memory until it commits. A commit sends the group's coordinator 1 number per partition: the offset of the next record to read, which is the last processed offset plus 1. After processing offsets 7 through 9 of partition 0, the consumer commits 10. The coordinator appends that number to an internal topic named __consumer_offsets, keyed by the group and the topic-partition, so the committed position belongs to the group rather than to the member that sent it. Any member that holds partition 0 later asks the coordinator for the group's committed offset and starts at 10.

The internal topic obeys everything the earlier lessons established about topics: it has partitions and replicas, and each commit is appended to it. Each new commit for a group and partition is a new record with a newer offset number inside it, and the coordinator keeps the latest in memory to answer quickly. Because the replicas carry the topic, a committed offset survives the consumer that wrote it, the member that replaces it, and the broker that stored it.

The last row shows what a restart does. The consumer reads from the committed offset, so every record between the last commit and the crash is read a 2nd time. Offsets 10 through 12 were charged once before the crash and are charged again after it. Kafka lost nothing; the gap between the position in memory and the position committed turned into duplicates downstream. The smaller that gap, the fewer the repeats, which is why the loop in section 0 committed after every batch.

Commit by hand or on a timer

bootstrap.servers=broker-1:9092
group.id=billing
enable.auto.commit=true
auto.commit.interval.ms=5000

With enable.auto.commit=true, the default, the consumer commits on its own: during a poll, if auto.commit.interval.ms has elapsed since the last commit, 5 seconds by default, it commits the position its memory holds, and your loop never calls commit. The timer runs on the consumer's clock and takes no notice of whether your handler finished with the batch; what that means for a crash in the middle of one is the next lesson's first subject. With enable.auto.commit=false, nothing is committed until the loop calls commit, and the loop picks the moment: once a batch, once every 100 records, or right after a downstream write succeeds.

commit comes in 2 forms. The synchronous form blocks until the coordinator confirms, so the loop pauses for a round trip but knows the position was saved before the next poll. The asynchronous form sends the commit and returns at once, so the loop keeps polling and the confirmation, or the failure, arrives later. The Java client names them commitSync and commitAsync. A loop that commits once a batch usually affords the synchronous round trip; a loop that commits after every record usually cannot.

In the base run the consumer commits after each batch, crashes after processing its 2nd batch, and on restart reads that batch again: 1 batch of duplicate rows lands in the ledger. In the variant the loop never commits, so the restart finds no saved position for the group and the whole partition is read from the start, doubling every row it had written. The difference between the 2 runs is 1 line in the loop.

• 50
• 1

About this lesson

This section is part of the Consumer Groups: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.