The Producer API

The Producer API

The Producer API and the Send Path

produce() (Java: send()) only queues: it serializes the record, picks a partition, and appends it to that partition's batch in the record accumulator. A sender thread ships ready batches, one request per broker, and returns each answer as a delivery callback, which Python runs only inside poll() or flush().

The producer's send path: the application thread only queues; the sender thread talks to brokers

l0651_send.py: two events of order 7 and their delivery reportsPython

from confluent_kafka import Producer
producer = Producer({"bootstrap.servers": "localhost:33092", "client.id": "booknest-shop"})
def report(err, msg):                         # delivery callback, run by poll() or flush()
    print(f"failed: {err}" if err else f"key {msg.key().decode()} -> partition "
          f"{msg.partition()} offset {msg.offset()} ({msg.latency() * 1000:.1f} ms)")
for event in ('{"type":"order_placed","order_id":7}', '{"type":"order_paid","order_id":7}'):
    producer.produce("booknest.order-events", event, "7", on_delivery=report)  # only queues
producer.flush(10)                            # wait for every delivery report

Output

key 7 -> partition 0 offset 455 (9.6 ms)
key 7 -> partition 0 offset 456 (9.7 ms)

Both calls returned at once; flush() waited for the reports. Share one thread-safe producer per process.

MENU