Producing Messages with Python

A producer turns application events into Kafka records. This article uses the confluent-kafka library, which wraps the battle-tested C client librdkafka and is the common choice for production Python producers and consumers.

The one thing to internalise up front: produce() is asynchronous. It does not send a record and wait; it puts the record in an in-memory buffer and returns immediately. Delivery happens in the background, and you find out the result later. Almost every producer mistake comes from forgetting this.

Mental model#

event dict   |  json.dumps(...).encode("utf-8")value bytes  (+ optional key bytes)   |  producer.produce(topic, key, value, on_delivery=cb)in-memory send queue  (bounded by queue.buffering.max.messages)   |  a background thread batches and sends;   |  linger.ms and batch.size control how much it waits and groupsbroker  ->  appends to a partition, returns an ack   |  the background thread invokes your delivery callback   |  poll() / flush() is what lets those callbacks run on your thread

produce() only fails synchronously in one case: the buffer is full. Everything else — the broker was unreachable, the message was too large, the topic does not exist — is reported asynchronously, through the delivery callback.

Install and configure#

pip install confluent-kafka

The only required setting is the broker address:

from confluent_kafka import Producer
producer = Producer({    "bootstrap.servers": "localhost:9092",})

Use 127.0.0.1:9092 from your host and kafka:29092 from inside a container (see Running Kafka Locally). bootstrap.servers can list several brokers, comma-separated; the client only needs to reach one to discover the rest.

Build one producer per process and reuse it. Each producer holds connections, a background thread, and the send buffer; creating one per message is slow and defeats batching.

In real code, the broker address and other settings come from environment variables or configuration, not literals.

Sending a message#

produce() takes a topic and a value. The value must be bytes or str, so serialize structured data first:

import json
event = {"order_id": 101, "customer_id": 1, "amount": 1200.0}
producer.produce(    topic="order-events",    value=json.dumps(event).encode("utf-8"),)
producer.poll(0)

producer.poll(0) gives the client a moment to do background work and run any delivery callbacks that are ready. Call it after producing, and periodically in a long-running loop. Without it, callbacks never fire and internal timers do not advance.

Serialization and the message contract#

Kafka stores bytes. The producer and every consumer must agree on what those bytes mean. For JSON that agreement is informal but real:

• Field names and types are a contract. A consumer parsing event["amount"] as a float breaks if the producer starts sending it as a string.
• Additive changes are safe; removals and renames are not. Adding a new optional field does not break existing consumers. Removing or renaming one does.
• Encode consistently. json.dumps(...).encode("utf-8") on the way out, json.loads(msg.value().decode("utf-8")) on the way in.

For stricter guarantees, teams use Avro or Protobuf with a Schema Registry that enforces compatibility. That is out of scope here; the practical minimum is a documented JSON shape and discipline about changing it. See the contract discussion in Kafka in a Data Pipeline.

Keys and partitioning#

A record can carry a key. Kafka hashes the key to choose a partition, so all records with the same key land in the same partition and stay ordered relative to each other:

producer.produce(    topic="order-events",    key=str(event["customer_id"]).encode("utf-8"),    value=json.dumps(event).encode("utf-8"),)

Decisions:

• Use a key when order matters for an entity. All events for one order, or one customer, or one device should be keyed by that id so a consumer sees them in sequence.
• No key when order does not matter. Keyless records spread evenly across partitions, which maximises throughput.
• Avoid low-cardinality keys. Keying by country or a boolean funnels most traffic into one partition — a hot partition that becomes the bottleneck. Key by something with many distinct values.

The key is also stored with the record, so consumers and compacted topics can use it.

Delivery reports#

Pass on_delivery to learn the outcome of each record. The callback runs when poll() or flush() processes the result:

def on_delivery(err, msg):    if err is not None:        print(f"delivery failed for key={msg.key()}: {err}")    else:        print(f"delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")

producer.produce(    topic="order-events",    value=json.dumps(event).encode("utf-8"),    on_delivery=on_delivery,)producer.poll(0)

What you do in the callback depends on the application. For a data pipeline, at minimum log failures with enough context (the key, the error) to investigate. If a record must not be lost, keep it and retry it, or write it to a fallback store.

Flushing before exit#

flush() blocks until every buffered record has a delivery result, or the timeout passes. It returns the number still undelivered. Always call it before the process ends, or the last batch is silently dropped:

remaining = producer.flush(timeout=10)if remaining > 0:    print(f"{remaining} messages were not delivered")

A common structure is poll(0) inside the loop for steady progress, and one flush() at shutdown.

Reliability settings: acks, idempotence, ordering#

The defaults favour throughput. For data you care about:

Turning on enable.idempotence is the single highest-value change: it makes the producer safe to retry without creating duplicate records in a partition.

Throughput settings#

The client batches automatically. To batch more aggressively:

Producing a batch#

The client already batches on the wire. "Producing a batch" from your code means assembling a group of events, producing them all, and calling flush() once at the boundary instead of after each message — so the whole group is one retryable unit.

def build_order_batch(orders):    """Turn raw records into the message dicts to publish."""    batch = []    for order in orders:        batch.append({            "order_id": order["id"],            "customer_id": order["customer_id"],            "created_at": order["created_at"].isoformat(),            "amount": float(order["amount"]),        })    return batch

def produce_batch(producer, topic, batch):    for event in batch:        producer.produce(            topic=topic,            key=str(event["customer_id"]).encode("utf-8"),            value=json.dumps(event).encode("utf-8"),            on_delivery=on_delivery,        )        producer.poll(0)
    remaining = producer.flush(timeout=10)    if remaining > 0:        print(f"{remaining} of {len(batch)} messages not delivered")    return remaining

When reading from a stream rather than a finite list, accumulate up to a size and send whenever the batch fills:

BATCH_SIZE = 500batch = []
for order in stream_of_orders():    batch.append(order_to_event(order))    if len(batch) >= BATCH_SIZE:        produce_batch(producer, "order-events", batch)        batch.clear()
if batch:    produce_batch(producer, "order-events", batch)   # flush the tail

A complete producer#

import jsonfrom datetime import datetimefrom confluent_kafka import Producer
TOPIC = "order-events"

def on_delivery(err, msg):    if err is not None:        print(f"delivery failed: {err}")

def build_producer():    return Producer({        "bootstrap.servers": "localhost:9092",        "enable.idempotence": True,        "acks": "all",        "linger.ms": 10,        "compression.type": "lz4",    })

def build_order_batch(orders):    batch = []    for order in orders:        batch.append({            "order_id": order["id"],            "customer_id": order["customer_id"],            "created_at": order["created_at"].isoformat(),            "amount": float(order["amount"]),        })    return batch

def produce_batch(producer, batch):    for event in batch:        producer.produce(            topic=TOPIC,            key=str(event["customer_id"]).encode("utf-8"),            value=json.dumps(event).encode("utf-8"),            on_delivery=on_delivery,        )        producer.poll(0)
    remaining = producer.flush(timeout=10)    if remaining > 0:        print(f"{remaining} of {len(batch)} messages not delivered")    return remaining

if __name__ == "__main__":    raw_orders = [        {"id": 101, "customer_id": 1, "created_at": datetime(2026, 1, 4, 9, 12), "amount": 1200},        {"id": 102, "customer_id": 1, "created_at": datetime(2026, 1, 4, 10, 40), "amount": 850},        {"id": 103, "customer_id": 2, "created_at": datetime(2026, 1, 5, 14, 5), "amount": 640},    ]
    producer = build_producer()    produce_batch(producer, build_order_batch(raw_orders))

Troubleshooting#

Further reading#

• confluentinc/confluent-kafka-python — the library's source and API documentation, with producer and consumer examples and the full list of configuration properties.
• Продюсеры и консьюмеры с Apache Kafka в Python — a longer Russian-language walkthrough of building producers and consumers with confluent-kafka-python.

Common mistakes#

Not calling flush()#

Buffered messages are lost when the process exits. Flush before shutting down.

Producing a dict directly#

produce() needs bytes or str. Serialize with json.dumps and .encode().

Never calling poll()#

Delivery callbacks and internal timers do not run. Call poll(0) after producing, or in your loop.

A new Producer per message#

Creating a producer opens connections and a thread and defeats batching. Build one and reuse it.

Treating produce() as synchronous#

It returns after buffering, not after delivery. Confirmation comes through the delivery callback or flush().

Keying by a low-cardinality field#

key=country creates a hot partition. Key by an id with many distinct values, or use no key.

Quick reference#

See also#