Commit offsets yourself

The short version

A commit saves the position of a group in a partition. Turn off automatic commits and call commit after the work is done.

Details

Automatic commits run on a timer, so a crash can skip messages that were committed but never handled. Committing after the work means a crash can only cause a repeat. This is called at least once delivery, so make your handling safe to repeat.

Change the config of your consumer:

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "billing",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})

Then commit after each message is handled:

        print(msg.key().decode(), msg.value().decode(), msg.offset())
        consumer.commit(message=msg, asynchronous=False)

Try it yourself

Make both changes in consume.py, run it, then stop it with Ctrl+C and describe the group:

docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --describe --group billing

You should see LAG at 0 on every partition, because each message was committed after it was printed.

Explain it to the duck

Why is a repeat safer than a skip, and what does your code need to do to handle a repeat?

Tested with: Kafka 4.1.0 in Docker, Python 3.12 and confluent-kafka 2.11.