Produce and Consume Kafka Messages in Python

Apache Kafka is a distributed event streaming platform. It lets services publish and subscribe to streams of records reliably.

Python is a great language for Kafka clients. The kafka-python library gives you a clean API for producers and consumers.

This guide shows you how to produce and consume Kafka messages in Python. It covers setup, keyed messages, consumer groups, offsets, and error handling.

Prerequisites

You need Python 3.8 or newer and a running Kafka broker. A local Kafka instance or a managed service works fine.

Install the client inside a virtual environment. If you have not done this yet, follow a guide to install kafka-python in Python first.

You also need network access to the broker. The default plaintext port is 9092.

Understand the Core Concepts

A topic is a named stream of records. Producers write to topics, and consumers read from them.

Each topic is split into partitions. Partitions allow parallel processing and ordered delivery within a partition.

A producer sends records to a topic. A consumer subscribes to a topic and reads records.

A consumer group lets multiple consumers share the work. Kafka assigns partitions across group members automatically.

Messages with the same key go to the same partition. This guarantees ordering for related events.

Create a Producer

The KafkaProducer class sends records asynchronously. It buffers messages and flushes them in batches for efficiency.

Pass the broker address with bootstrap_servers. A value serializer converts Python values to bytes.

from kafka import KafkaProducer

# Create a producer connected to a local broker
producer = KafkaProducer(
    bootstrap_servers=["localhost:9092"],
    value_serializer=lambda v: v.encode("utf-8"),
)

# Send a message asynchronously
future = producer.send("events", "user signed up")

# Block until the send completes
metadata = future.get(timeout=10)
print(f"Sent to partition {metadata.partition} at offset {metadata.offset}")

# Flush pending messages and close
producer.flush()
producer.close()

Sent to partition 0 at offset 7

The send() method returns a future. Calling get() on that future waits for broker acknowledgement.

The flush() method forces all buffered messages to the broker. Always call it before closing the producer.

Send Keyed Messages

Keys control partitioning. Messages with the same key land in the same partition.

Use keys when order matters, such as per-user or per-order events.

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=["localhost:9092"])

# Messages with the same key go to the same partition
producer.send("orders", key=b"user-42", value=b'{"order_id": 1001}')
producer.send("orders", key=b"user-42", value=b'{"order_id": 1002}')

producer.flush()
print("Keyed messages sent.")
producer.close()

Keyed messages sent.

Kafka hashes the key to pick a partition. Identical keys always map to the same partition.

Create a Consumer

The KafkaConsumer class reads records from topics. Join a consumer group with a group_id.

Set auto_offset_reset="earliest" so a new group reads from the start.

from kafka import KafkaConsumer

# Create a consumer in a group
consumer = KafkaConsumer(
    "events",
    bootstrap_servers=["localhost:9092"],
    group_id="event-processors",
    auto_offset_reset="earliest",
)

# Iterate over messages
for message in consumer:
    print(f"{message.topic}:{message.partition}:{message.offset} -> {message.value.decode()}")

events:0:7 -> user signed up

The consumer iterator blocks when no messages are available. It returns records as soon as they arrive.

Use subscribe() to join topics dynamically instead of passing them to the constructor.

consumer = KafkaConsumer(
    bootstrap_servers=["localhost:9092"],
    group_id="event-processors",
    auto_offset_reset="earliest",
)
consumer.subscribe(["events", "orders"])

Poll for Messages

The poll() method fetches messages without blocking indefinitely. It returns a dictionary keyed by topic partition.

This pattern fits applications that need to do other work between fetches.

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    bootstrap_servers=["localhost:9092"],
    group_id="poll-group",
    auto_offset_reset="earliest",
)
consumer.subscribe(["events"])

while True:
    batch = consumer.poll(timeout_ms=1000)
    if not batch:
        print("No new messages.")
        continue
    for topic_partition, messages in batch.items():
        for message in messages:
            print(f"Received: {message.value.decode()}")

Received: user signed up
No new messages.

The timeout_ms argument controls how long the call waits. A short timeout keeps your loop responsive.

Manage Offsets Manually

Offsets track the position of each consumer in a partition. Committing offsets after processing gives you control over delivery guarantees.

Set enable_auto_commit=False and commit manually. This avoids message loss when a consumer crashes mid-handler.

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "events",
    bootstrap_servers=["localhost:9092"],
    group_id="manual-commit-group",
    enable_auto_commit=False,
    auto_offset_reset="earliest",
)

for message in consumer:
    process(message)
    consumer.commit()  # Commit after successful processing

The commit() method records the current offset for the consumer group. Only commit after your work succeeds.

Handle Producer Errors

Kafka operations can fail. Network issues, broker restarts, and timeouts are common.

Wrap sends in a try block and catch KafkaError. Decide whether to retry or log the failure.

from kafka import KafkaProducer
from kafka.errors import KafkaError

producer = KafkaProducer(bootstrap_servers=["localhost:9092"])

try:
    future = producer.send("events", b"important message")
    metadata = future.get(timeout=10)
    print(f"Delivered to offset {metadata.offset}")
except KafkaError as exc:
    print(f"Send failed: {exc}")
finally:
    producer.flush()
    producer.close()

Delivered to offset 8

The future.get() call surfaces delivery failures that would otherwise be silent. Always wait on the future when the message matters.

For production monitoring, you can install sentry-sdk in Python and capture Kafka exceptions automatically.

Handle Consumer Errors

Consumers can fail during connection or while processing messages. Catch specific exceptions and retry cleanly.

from kafka import KafkaConsumer
from kafka.errors import KafkaError
import time

while True:
    try:
        consumer = KafkaConsumer(
            "events",
            bootstrap_servers=["localhost:9092"],
            group_id="resilient-group",
            auto_offset_reset="earliest",
        )
        for message in consumer:
            process(message)
    except KafkaError as exc:
        print(f"Consumer error: {exc}. Reconnecting in 5 seconds.")
        time.sleep(5)

This loop reconnects after connection failures. It keeps the consumer running without manual intervention.

Troubleshoot Connection Errors

The NoBrokersAvailable error means the client cannot reach any broker. It is the most common Kafka connection error.

Check the broker status first. Confirm that the Kafka process is running and that port 9092 is open.

# Test TCP connectivity to the broker
nc -zv localhost 9092

If the connection fails, the broker is down or a firewall is blocking traffic. Fix the network path before retrying.

In Docker or Kubernetes, the advertised listener address must be reachable from your client. A broker reachable inside the cluster may not be reachable from your laptop.

Verify the bootstrap_servers address. A typo in the hostname or port causes this error immediately.

Best Practices for Clean Messaging

Reuse producer and consumer instances. Creating a new client per message wastes resources and slows down your application.

Use keys for ordering guarantees. Messages with the same key always land in the same partition.

Set enable_auto_commit=False for critical workloads. Manual commits give you control over delivery guarantees.

Keep processing fast. Offload heavy work to background workers or a separate queue.

If your consumer writes results to a database, you can install mysqlclient in Python to persist processed events from the same worker.

Monitor consumer lag and producer failures. A quiet topic often means a broken consumer, not an empty queue.

Test failure paths locally. Stop the broker and confirm that your retry logic reconnects cleanly.

Pin the kafka-python version in your dependency file. Kafka protocol support changes between releases.

Conclusion

Producing and consuming Kafka messages in Python is straightforward with kafka-python. Create a producer, send records, and flush before closing.

Create a consumer in a group, read messages, and commit offsets manually for reliable delivery.

Handle errors with KafkaError and add retry logic for connection failures.

With these patterns, your Python services can exchange Kafka messages reliably and safely at any scale.