How to Produce and Consume Kafka Messages in Python

The producer

KafkaProducer connects to the broker, serializes your dicts to JSON bytes, and publishes to a named topic. acks="all" means every in-sync replica must acknowledge before the send returns - the right default for anything important.

import json, os, random
from kafka import KafkaProducer

BROKER = os.environ["KAFKA_BROKER"]
TOPIC  = "orders"

producer = KafkaProducer(
    bootstrap_servers=BROKER,
    value_serializer=lambda v: json.dumps(v).encode(),
    acks="all",
    retries=3,
)

for i in range(1, 11):
    event = {
        "order_id": i,
        "customer_id": random.randint(1, 100),
        "amount": round(random.uniform(1, 500), 2),
    }
    # .get() makes it synchronous - raises on broker error
    producer.send(TOPIC, value=event).get(timeout=10)
    print(f"sent {event}")

producer.flush()   # drain any buffered messages
producer.close()

The value_serializer runs before each send, so you pass plain dicts and Kafka gets bytes. producer.send(...).get(timeout=10) blocks until the broker confirms - slower than fire-and-forget but safe for order events where you can't afford silent drops.

The consumer

KafkaConsumer subscribes to the topic and yields messages as an iterator. auto_offset_reset="earliest" means start from the beginning of the topic if this consumer group has never committed an offset - essential for reading events the producer already wrote before the consumer started.

import json, os
from kafka import KafkaConsumer

BROKER = os.environ["KAFKA_BROKER"]
TOPIC  = "orders"

consumer = KafkaConsumer(
    TOPIC,
    bootstrap_servers=BROKER,
    group_id="orders-consumer",
    auto_offset_reset="earliest",
    enable_auto_commit=True,
    value_deserializer=lambda b: json.loads(b),
    consumer_timeout_ms=10000,   # iterator stops if no new messages for 10s
)

received = []
for msg in consumer:
    received.append(msg.value)
    print(f"got {msg.value}")
    if len(received) >= 10:
        break

consumer.close()

consumer_timeout_ms turns the iterator into a finite loop - without it the for msg in consumer blocks forever waiting for more messages.

Why Redpanda works as a drop-in

Redpanda speaks the Kafka wire protocol on port 9092. Your kafka-python client connects the same way, uses the same API, and the same topic/group semantics apply. Redpanda is faster to start (single binary, no JVM) and is a common local dev / CI stand-in for Kafka.

Producer run order matters

Run producer.py first to publish the 10 events, then consumer.py. Because auto_offset_reset="earliest" is set, the consumer will rewind to offset 0 on first run and pick up all existing messages - even if it starts after the producer finishes.