What's the difference between at-most-once and at-least-once consumption?

At-most-once commits before processing (a crash loses messages); at-least-once commits after processing (a crash re-delivers some messages).

Why make Kafka consumers idempotent?

Messages can be delivered more than once after crashes and rebalances; idempotent handling makes duplicates harmless.

Why disable auto-commit in many services?

Auto-commit commits on a timer, possibly before processing finished; explicit commits after processing avoid losing messages.

Offset Commits and Delivery Semantics — Kafka in Production

What you'll build

A consumer that crashes halfway through a batch. With commits after processing it re-reads the unfinished messages (at-least-once), and an idempotent handler (deduplication by event ID) turns the redelivery into exactly-once effects.

The idea

The committed offset is where a group resumes after a restart.
At-most-once: commit before processing; a crash loses messages.
At-least-once: commit after processing; a crash re-delivers messages that were processed but not committed. This is the usual choice.
Make handlers idempotent (deduplicate by an event ID, use upserts or conditional updates) so redelivery is harmless.
Turn off auto-commit (enable.auto.commit=false) and commit explicitly when you control processing.

The code

Crash, redelivery and idempotent processing

import os
from confluent_kafka import Consumer, Producer, TopicPartition
from confluent_kafka.admin import AdminClient, NewTopic

BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]
admin = AdminClient({"bootstrap.servers": BOOTSTRAP})
admin.create_topics([NewTopic("credits", num_partitions=1, replication_factor=1)])["credits"].result()
producer = Producer({"bootstrap.servers": BOOTSTRAP, "enable.idempotence": True})
for i in range(1, 7):
    producer.produce("credits", key="acc-1", value=f"evt-{i}:10")
producer.flush()

balance, processed_ids = {"acc-1": 0}, set()
def handle(event_id, amount):
    if event_id in processed_ids:               # idempotency: a redelivered event changes nothing
        return "skipped duplicate"
    processed_ids.add(event_id)
    balance["acc-1"] += amount
    return "applied"

def run(crash_after=None):
    consumer = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "billing",
                         "auto.offset.reset": "earliest", "enable.auto.commit": False})
    consumer.subscribe(["credits"])
    handled = 0
    while True:
        msg = consumer.poll(5.0)
        if msg is None:
            break
        event_id, amount = msg.value().decode().split(":")
        print(f"  offset {msg.offset()}: {event_id} {handle(event_id, int(amount))}")
        handled += 1
        if crash_after and handled == crash_after:
            print("  crash before committing!")
            consumer.close()                     # nothing committed after offset 1
            return
        if handled % 2 == 0:
            consumer.commit(asynchronous=False)  # commit after processing (every 2 messages)
    consumer.close()

print("first run:")
run(crash_after=3)
print("after restart:")
run()
print("balance:", balance["acc-1"], "(6 credits of 10)")
admin.delete_topics(["credits"])["credits"].result()
admin.delete_consumer_groups(["billing"])["billing"].result()

import os
from confluent_kafka import Consumer, Producer, TopicPartition
from confluent_kafka.admin import AdminClient, NewTopic

BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]
admin = AdminClient({"bootstrap.servers": BOOTSTRAP})
admin.create_topics([NewTopic("credits", num_partitions=1, replication_factor=1)])["credits"].result()
producer = Producer({"bootstrap.servers": BOOTSTRAP, "enable.idempotence": True})
for i in range(1, 7):
    producer.produce("credits", key="acc-1", value=f"evt-{i}:10")
producer.flush()

balance, processed_ids = {"acc-1": 0}, set()
def handle(event_id, amount):
    if event_id in processed_ids:               # idempotency: a redelivered event changes nothing
        return "skipped duplicate"
    processed_ids.add(event_id)
    balance["acc-1"] += amount
    return "applied"

def run(crash_after=None):
    consumer = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "billing",
                         "auto.offset.reset": "earliest", "enable.auto.commit": False})
    consumer.subscribe(["credits"])
    handled = 0
    while True:
        msg = consumer.poll(5.0)
        if msg is None:
            break
        event_id, amount = msg.value().decode().split(":")
        print(f"  offset {msg.offset()}: {event_id} {handle(event_id, int(amount))}")
        handled += 1
        if crash_after and handled == crash_after:
            print("  crash before committing!")
            consumer.close()                     # nothing committed after offset 1
            return
        if handled % 2 == 0:
            consumer.commit(asynchronous=False)  # commit after processing (every 2 messages)
    consumer.close()

print("first run:")
run(crash_after=3)
print("after restart:")
run()
print("balance:", balance["acc-1"], "(6 credits of 10)")
admin.delete_topics(["credits"])["credits"].result()
admin.delete_consumer_groups(["billing"])["billing"].result()

Output

first run:
  offset 0: evt-1 applied
  offset 1: evt-2 applied
  offset 2: evt-3 applied
  crash before committing!
after restart:
  offset 2: evt-3 skipped duplicate
  offset 3: evt-4 applied
  offset 4: evt-5 applied
  offset 5: evt-6 applied
balance: 60 (6 credits of 10)

Run this on your computer — see “Get Started” for setup.

Offset 2 was processed but not committed before the crash, so it was delivered again; the idempotent handler skipped it and the balance is correct.

Common mistakes

Auto-commit with asynchronous processing: offsets get committed for work that hasn't finished, losing messages on a crash.
Assuming "exactly once" without idempotent side effects; redelivery always happens eventually (crashes, rebalances).
Storing the deduplication set in memory only; in production use the database (unique constraint on event ID) in the same transaction as the effect.

Key points

Commit after processing for at-least-once delivery.
Expect redelivery and make handlers idempotent.
Keep deduplication durable, alongside the side effect.

Further reading: Message delivery semantics · Consumer configs: enable.auto.commit

What you'll build

The idea

The code

Common mistakes

Key points

Practice questions