Install kafka-python in Python

kafka-python is a pure Python client for Apache Kafka. It lets your applications produce and consume messages without a JVM or complex native dependencies.

This guide covers installation, basic producer and consumer setup, error handling, and troubleshooting. It is written for beginners who want a clean and working configuration.

Every section includes practical code and clear explanations. You can follow along and adapt the examples to your own project.

Prerequisites

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

Check your Python version before you start. Recent releases of kafka-python require Python 3.8 and later.

python --version
python3 --version

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

Set Up a Virtual Environment

Always isolate project dependencies. A virtual environment keeps packages separate and avoids version conflicts.

# Create a virtual environment
python -m venv kafka-env

# Activate on Linux or macOS
source kafka-env/bin/activate

# Activate on Windows
kafka-env\Scripts\activate

Your prompt should change after activation. Confirm the interpreter path with which python or where python.

Install kafka-python with pip

The package is a pure Python library. Installation is a single pip command with no external build dependencies.

pip install --upgrade kafka-python

The --upgrade flag ensures you get the latest stable release. This matters because the client evolves with Kafka protocol changes.

Pin the version in your dependency file for reproducible deployments.

pip freeze | grep kafka-python >> requirements.txt

Install the Optional crc32c Dependency

kafka-python calculates record checksums in pure Python by default. That calculation is CPU intensive.

Install the optional crc32c package to offload checksum work to an optimized C library.

pip install crc32c

This is highly recommended for production workloads. It reduces CPU cost for every record the client processes.

Create a Kafka Producer

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

Pass the broker address with bootstrap_servers. Then call producer.send() to publish a message.

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("my-topic", "Hello from kafka-python")

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

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

Sent to my-topic partition 0 offset 42

The future.get() method blocks for a synchronous result. Without it, sends happen in the background.

The producer.flush() method waits for all pending messages to reach the broker. Call it before closing.

Send Messages with Keys

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

Use a key 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 are hashed 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 Kafka Consumer

The KafkaConsumer class reads messages from topics. It supports consumer groups for parallel processing.

Pass a group_id to join a consumer group. Kafka assigns partitions across group members automatically.

from kafka import KafkaConsumer

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

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

my-topic:0:42 key=None value=b'Hello from kafka-python'

The auto_offset_reset="earliest" setting makes a new group read from the beginning. Without it, the consumer starts at the latest offset.

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

Use poll() for Non-Blocking Consumption

The poll() method fetches messages without blocking indefinitely. It returns a dictionary of messages 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(["my-topic"])

# Poll for messages with a timeout
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}")

Received: b'Hello from kafka-python'
No new messages.

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

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("my-topic", b"important message")
    record_metadata = future.get(timeout=10)
    print(f"Delivered to offset {record_metadata.offset}")
except KafkaError as exc:
    print(f"Send failed: {exc}")
finally:
    producer.flush()
    producer.close()

Delivered to offset 43

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. This helps you spot broken producers before data loss occurs.

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(
            "my-topic",
            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.

Commit offsets only after successful processing. Auto-commit can lose messages if your process crashes mid-handler.

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

for message in consumer:
    process(message)
    consumer.commit()

The consumer.commit() method records the offset after each successful message. This gives you at-least-once delivery guarantees.

Troubleshoot NoBrokersAvailable

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.

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

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.

On newer versions of kafka-python, NoBrokersAvailableError has been removed. Connection failures now raise KafkaTimeoutError during bootstrap or specific authentication errors at send time. Check your client version if the error message does not match.

Troubleshoot Authentication Errors

Secured clusters require SASL or SSL configuration. Missing credentials appear as connection failures.

For SASL/PLAIN, pass the username and password to the client.

from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=["kafka.example.com:9093"],
    security_protocol="SASL_SSL",
    sasl_mechanism="PLAIN",
    sasl_plain_username="myuser",
    sasl_plain_password="mypassword",
    ssl_cafile="/path/to/ca.crt",
)

For SSL-only clusters, provide the CA certificate file.

producer = KafkaProducer(
    bootstrap_servers=["kafka.example.com:9093"],
    security_protocol="SSL",
    ssl_cafile="/path/to/ca.crt",
)

Confirm the certificate paths and credentials before debugging further. A wrong path produces an SSL handshake error, not a clear authentication message.

Best Practices for a Clean Setup

Pin the kafka-python version in requirements.txt. Kafka protocol support changes between releases, so reproducibility matters.

pip freeze | grep kafka-python >> requirements.txt

Install the optional crc32c package in production. It reduces CPU usage for checksum calculation.

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

Use consumer groups for parallel processing. Kafka assigns partitions automatically and balances load across workers.

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

If you process database records after consuming Kafka messages, you can install mysqlclient in Python to write results from the same consumer. Keep the callback fast and offload heavy work to background workers.

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.

Conclusion

Installing kafka-python in Python is quick. Create a virtual environment, install the package with pip, and add the optional crc32c dependency for performance.

Use KafkaProducer to send messages and KafkaConsumer to read them. Handle errors with KafkaError and commit offsets manually for reliable delivery.

Troubleshoot connection issues by checking broker reachability and authentication settings.

With these steps, your Python applications can produce and consume Kafka messages reliably at any scale.