Reliable Producers: acks, Idempotence and Delivery Reports

What you'll build

A payment-events producer configured for durability: acks=all, idempotence on, delivery callbacks that record the partition and offset of every message (or the error), and a flush before exiting so nothing is left in the buffer.

The idea

• produce() is asynchronous: messages go into a buffer and are sent in batches. Delivery results arrive in callbacks during poll()/flush().
• acks=all: the leader waits for all in-sync replicas. With topic setting min.insync.replicas=2 (and replication factor 3), an acknowledged write survives a broker failure.
• enable.idempotence=true prevents duplicates from producer retries and keeps per-partition order (it's the default in modern clients, but set it explicitly).
• linger.ms and batch.size trade a little latency for much higher throughput.

The code

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

BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]
admin = AdminClient({"bootstrap.servers": BOOTSTRAP})
admin.create_topics([NewTopic("payments", num_partitions=3, replication_factor=1)])["payments"].result()

producer = Producer({
    "bootstrap.servers": BOOTSTRAP,
    "acks": "all",
    "enable.idempotence": True,
    "linger.ms": 20,               # wait up to 20 ms to build bigger batches
    "compression.type": "zstd",
    "client.id": "payment-service",
})

delivered, failed = [], []
def on_delivery(err, msg):
    if err:
        failed.append(str(err))
    else:
        delivered.append((msg.key().decode(), msg.partition(), msg.offset()))

for i, account in enumerate(["acc-1", "acc-2", "acc-1", "acc-3", "acc-1"]):
    producer.produce("payments", key=account, value=f'{{"payment": {i}, "amount": {10 * (i + 1)}}}', on_delivery=on_delivery)
    producer.poll(0)               # serve delivery callbacks without blocking

remaining = producer.flush(10)     # block until everything is delivered (or 10 s pass)
print("not delivered:", remaining, "| failures:", failed)
for key, partition, offset in sorted(delivered):
    print(f"{key} -> partition {partition}, offset {offset}")
admin.delete_topics(["payments"])["payments"].result()

not delivered: 0 | failures: []
acc-1 -> partition 2, offset 0
acc-1 -> partition 2, offset 1
acc-1 -> partition 2, offset 3
acc-2 -> partition 0, offset 0
acc-3 -> partition 2, offset 2

All three acc-1 payments went to the same partition with increasing offsets, preserving their order (another key that hashes to the same partition fills the gap).

Common mistakes

• Never calling poll() or flush(), so delivery callbacks never run and errors go unnoticed.
• Exiting without flush(), losing buffered messages.
• acks=1 (or 0) for important data: a leader failure right after acknowledging loses the write.
• Disabling idempotence and enabling retries, which can duplicate and reorder messages.

Key points

• produce() is asynchronous; check delivery reports and flush on shutdown.
• acks=all with min.insync.replicas makes acknowledged writes durable.
• Idempotence prevents retry duplicates and preserves order.

Further reading: Producer configs · confluent-kafka Python client