Producing to Kafka from Python

The console producer is fine for poking at a topic. Real producers run inside applications, and the interesting behaviour — batching, acknowledgement, failure handling — is only visible from code.

Installing the client

pip install confluent-kafka

Successfully installed confluent-kafka-2.6.1

The smallest producer that works

from confluent_kafka import Producer

producer = Producer({"bootstrap.servers": "localhost:9092"})

producer.produce("orders", key="order-1", value="espresso")
producer.flush()

print("sent")

$ python produce_one.py
sent

That works, but it tells you nothing. produce() returned immediately without contacting the broker at all — it appended the record to an in-memory queue and handed control back. The actual network send happened during flush().

If you delete the flush() line, the script still prints sent and exits — and the record never reaches Kafka. There is no error, because nothing failed; the process just ended before the background thread got to work. This is the single most common way to lose data with this client.

Seeing the acknowledgement

To learn what actually happened to a record, pass a delivery callback:

from confluent_kafka import Producer

def delivered(err, msg):
    if err is not None:
        print(f"FAILED: {err}")
    else:
        print(f"ok  topic={msg.topic()} partition={msg.partition()} offset={msg.offset()}")

producer = Producer({"bootstrap.servers": "localhost:9092"})

for drink in ["espresso", "cortado", "flat white"]:
    producer.produce("orders", key="counter", value=drink, callback=delivered)

producer.flush()

$ python produce_callbacks.py
ok  topic=orders partition=0 offset=4
ok  topic=orders partition=0 offset=5
ok  topic=orders partition=0 offset=6

Now the broker’s answer is visible: each record was assigned a partition and an offset. All three went to partition 0 because they share the key counter.

Notice all three callbacks fired at once, after the loop finished. The client runs callbacks on the thread that calls poll() or flush(), so in a tight loop they queue up until you give the client a chance to run them.

Producing continuously

A long-running producer should call poll(0) on each iteration. That serves the callback queue without blocking, so failures surface immediately rather than at shutdown.

import time
from confluent_kafka import Producer

def delivered(err, msg):
    if err:
        print(f"FAILED offset=? {err}")
    else:
        print(f"ok  partition={msg.partition()} offset={msg.offset()}")

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "acks": "all",              # wait for all in-sync replicas
    "linger.ms": 5,             # wait up to 5ms to fill a batch
    "compression.type": "zstd",
})

for i in range(5):
    producer.produce("orders", key=f"user-{i % 2}", value=f"event-{i}",
                     callback=delivered)
    producer.poll(0)            # serve callbacks, do not block
    time.sleep(0.2)

remaining = producer.flush(timeout=10)
print(f"unflushed records: {remaining}")

$ python produce_loop.py
ok  partition=0 offset=7
ok  partition=2 offset=5
ok  partition=0 offset=8
ok  partition=2 offset=6
ok  partition=0 offset=9
unflushed records: 0

The callbacks now interleave with production instead of arriving in a clump. user-0 and user-1 split across two partitions and each key stays on its own.

flush() returns the number of records still in the queue when it gave up. 0 means everything was acknowledged. A non-zero return after a timeout is your signal that records were not delivered — check it rather than assuming.

What happens when the broker is gone

Stop the broker (Ctrl+C in its terminal) and run the loop again:

$ python produce_loop.py
%3|1771059612.418|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)
FAILED offset=? Local: Message timed out
FAILED offset=? Local: Message timed out
FAILED offset=? Local: Message timed out
FAILED offset=? Local: Message timed out
FAILED offset=? Local: Message timed out
unflushed records: 0

Two useful things here. The %3|...|FAIL| line comes from librdkafka’s internal logger, not your callback — the client keeps retrying in the background and reports connection trouble as it goes. Your callback only fires once the record finally gives up, after message.timeout.ms (default 300 seconds, so this run had it lowered).

The records were not silently dropped: every one produced an error callback. A producer that ignores the err argument turns a total outage into a silent data loss, which is why the callback is the first thing to wire up, not the last.

Delivery configuration that matters

Modern defaults are good. The two you almost always set explicitly are linger.ms (for throughput) and compression.type (for cost).

Practice

producer.produce("does-not-exist", value="hello", callback=delivered)
producer.flush()

ok  partition=0 offset=0

import time
from confluent_kafka import Producer

def run(linger):
    p = Producer({"bootstrap.servers": "localhost:9092", "linger.ms": linger})
    start = time.perf_counter()
    for i in range(10_000):
        p.produce("bench", value=f"record-{i}")
        p.poll(0)
    p.flush()
    return time.perf_counter() - start

print(f"linger.ms=0   {run(0):.2f}s")
print(f"linger.ms=50  {run(50):.2f}s")

linger.ms=0   1.94s
linger.ms=50  0.61s

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "message.timeout.ms": 3000,
})

FAILED offset=? Local: Message timed out

ok  partition=0 offset=10
ok  partition=0 offset=11

Next: reading those records back from code, and how consumer groups divide the work.