TopolearnWhen you commit relative to processing decides whether a crash loses records or processes some twice.
A committed offset means "this group has handled everything before this position". After a crash or rebalance, consumption resumes at the committed offset. Its timing defines your delivery guarantee:
- commit before processing: a crash after the commit skips the record: at most once
- commit after processing: a crash before the commit replays the record: at least once
Auto-commit (enable.auto.commit=true) commits in the background on a timer. In librdkafka it commits offsets of messages already handed to your code, so a crash can lose in-flight work or replay up to a few seconds of records.
For at-least-once, disable auto-commit and commit after processing: consumer.commit(message=msg, asynchronous=False) per record is simplest but slow. Better: commit every N records or seconds, or commit asynchronously and do a final synchronous commit on shutdown. An alternative in librdkafka is enable.auto.offset.store=false plus store_offsets(msg) after processing, keeping auto-commit for the actual commits.
Note the committed value is the next offset to read (last processed + 1); the clients handle this when you pass the message.
At-least-once means duplicates are possible, so make processing idempotent: upserts keyed by event id, or deduplication tables. True exactly-once within Kafka requires transactions.
Example
from confluent_kafka import Consumer
def handle(msg):
print("processed", msg.offset()) # e.g. write to a database (idempotently)
consumer = Consumer({
"bootstrap.servers": "localhost:9092",
"group.id": "billing",
"enable.auto.commit": False, # we decide when work is done
"auto.offset.reset": "earliest",
})
consumer.subscribe(["orders"])
done = 0
try:
while done < 1000:
msg = consumer.poll(1.0)
if msg is None or msg.error():
continue
handle(msg) # 1. process
done += 1
if done % 100 == 0:
consumer.commit(asynchronous=False) # 2. commit current positions
finally:
consumer.commit(asynchronous=False) # 3. final commit on shutdown
consumer.close()Remember
- Commit after processing = at least once; before = at most once
- Disable auto-commit for control, commit in batches
- The committed offset is the next one to read
- At-least-once needs idempotent processing
Watch out for
- Committing with auto-commit while processing asynchronously in other threads, losing records on crash.