Demo Kafka Consumers

Welcome back. In this lesson you’ll learn how to write a Kafka consumer in Python. We’ll cover:

Creating a topic with multiple partitions
Producing messages with a Python producer (kafka-python)
Implementing a simple Python consumer that reads and prints messages
Real-world considerations for running consumers in production

Prerequisites

A running Kafka broker reachable at localhost:9092
Kafka command-line tools available (usually in ~/kafka/bin or /opt/kafka/bin)
Python 3 installed
Basic familiarity with Kafka topics, partitions, and consumer groups

1) Create the topic

To consume messages you first need a topic and some messages in it. If you run the topic creation command from the wrong working directory, you may see an error like:

root@kafka_host ~  ./kafka-topics.sh --create --topic multi-partition-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
-bash: ./kafka-topics.sh: No such file or directory

This indicates the kafka-topics.sh script is not found in the current folder. Kafka CLI tools live in the Kafka installation bin directory (for example ~/kafka/bin or /opt/kafka/bin). Change to that directory and run the command again:

cd ~/kafka/bin
./kafka-topics.sh --create --topic multi-partition-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Expected successful output:

Created topic multi-partition-topic.

If you are unsure where Kafka is installed, consult environment documentation or search for kafka-topics.sh with find / -name kafka-topics.sh 2>/dev/null.

2) Produce messages to the topic (Python producer)

Install a Python virtual environment and the Kafka client library kafka-python:

sudo apt update
sudo apt install -y python3-venv        # if not already installed
python3 -m venv kafka-env
source kafka-env/bin/activate
pip install kafka-python

Create a producer script (e.g., kafka_producer_example.py) with the following content:

# kafka_producer_example.py
import random
import logging
from kafka import KafkaProducer

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Kafka configuration
producer = KafkaProducer(bootstrap_servers='localhost:9092')

# Example coffee shop data
coffee_shops = [
    {"name": "Espresso Bliss", "location": "Downtown", "rating": 4.5},
    {"name": "Cappuccino Corner", "location": "Uptown", "rating": 4.2},
    {"name": "Latte Lounge", "location": "Suburbs", "rating": 4.8},
    {"name": "Mocha Magic", "location": "City Center", "rating": 4.6},
    {"name": "Coffee Haven", "location": "East Side", "rating": 4.3},
]

# Generate and send 10 messages
for i in range(10):
    coffee_shop = random.choice(coffee_shops)
    message = f"Coffee Shop: {coffee_shop['name']}, Location: {coffee_shop['location']}, Rating: {coffee_shop['rating']}"

    try:
        # Send message (key and value must be bytes)
        future = producer.send('multi-partition-topic', key=str(i).encode('utf-8'), value=message.encode('utf-8'))
        # Block until the message is sent (or an exception)
        record_metadata = future.get(timeout=10)
        logging.info(f"Message delivered to {record_metadata.topic} [{record_metadata.partition}] at offset {record_metadata.offset}")
    except Exception as e:
        logging.error(f"Message delivery failed: {e}")

producer.flush()
producer.close()
logging.info("Finished sending messages")

Run the producer:

python3 kafka_producer_example.py

You should see logs showing connections and message delivery, for example:

2025-04-13 02:58:42,749 - INFO - Message delivered to multi-partition-topic [0] at offset 0
2025-04-13 02:58:42,749 - INFO - Message delivered to multi-partition-topic [1] at offset 0
2025-04-13 02:58:42,750 - INFO - Message delivered to multi-partition-topic [2] at offset 0
2025-04-13 02:58:42,750 - INFO - Finished sending messages

You can also verify messages with a Kafka UI (if available) by opening the topic and inspecting messages per partition.

3) Write the Python consumer

Create kafka_consumer.py and add the following consumer code. This consumer subscribes to multi-partition-topic, reads from the earliest offset when there is no committed offset, and prints message contents with partition and offset metadata:

# kafka_consumer.py
from kafka import KafkaConsumer

# Kafka Consumer configuration
consumer = KafkaConsumer(
    'multi-partition-topic',
    bootstrap_servers='localhost:9092',
    group_id='partition-checker-group',  # consumer group id
    auto_offset_reset='earliest'         # start reading from the earliest available message
)

try:
    print("Listening for messages...")
    for message in consumer:
        # message.value is bytes, decode to string
        print(f"Message received: {message.value.decode('utf-8')}")
        print(f"Partition: {message.partition}, Offset: {message.offset}")
except KeyboardInterrupt:
    print("Exiting...")
finally:
    consumer.close()

Key consumer settings explained:

Setting	Purpose	Example
`topic`	Topic(s) to subscribe to	`'multi-partition-topic'`
`bootstrap_servers`	Kafka broker addresses to connect to	`'localhost:9092'`
`group_id`	Consumer group identifier; consumers in the same group share partitions	`'partition-checker-group'`
`auto_offset_reset`	Where to start when no committed offset exists	`'earliest'` or `'latest'`

Run the consumer:

python3 kafka_consumer.py

Expected output (example):

Listening for messages...
Message received: Coffee Shop: Cappuccino Corner, Location: Uptown, Rating: 4.2
Partition: 2, Offset: 0
Message received: Coffee Shop: Latte Lounge, Location: Suburbs, Rating: 4.8
Partition: 0, Offset: 1
...

When you stop the consumer with Ctrl+C it exits cleanly and closes the connection.

If you omit group_id, Kafka treats the consumer as a new, distinct consumer; it may receive duplicates or begin from offsets determined by auto_offset_reset. Both bootstrap_servers and topic are required.

4) Real-world considerations

The example prints messages to stdout for demonstration. In production you would typically:

Parse the message payload (JSON, Avro, Protobuf, etc.)
Apply business logic and transformations
Persist results to a database or forward processed events to another topic
Add robust error handling, retries, backoff, and dead-letter queues
Instrument metrics and tracing for monitoring and debugging
Gracefully handle shutdown signals and rebalances to avoid duplicate processing

Useful patterns and topics to research:

Exactly-once vs at-least-once processing
Partitioning and keys (how message keys affect partition assignment)
Consumer rebalances and max.poll.interval.ms
Offset committing strategies (automatic vs manual commits)

Links and References

Kafka documentation: https://kafka.apache.org/documentation/
kafka-python (PyPI): https://pypi.org/project/kafka-python/
Kafka consumer groups and partitioning: https://kafka.apache.org/documentation/#consumerapi
Kubernetes for running consumers in production: https://learn.kodekloud.com/user/courses/kubernetes-for-the-absolute-beginners-hands-on-tutorial

Thanks for reading — that’s it for this lesson on building a Python Kafka consumer.

Foundations of Event Streaming

Building Blocks of Kafka

Kafka Producers & Consumers: The Message Flow

Deep Dive into Kafka: Beyond the Basics

Confluent Kafka and Its Offerings

Kafka Connect: Effortless Data Pipelines

Project: Building an Event-Driven System

Introduction

Demo Kafka Consumers

1) Create the topic

2) Produce messages to the topic (Python producer)

3) Write the Python consumer

4) Real-world considerations

Watch Video

Practice Lab

​1) Create the topic

​2) Produce messages to the topic (Python producer)

​3) Write the Python consumer

​4) Real-world considerations

Watch Video

Practice Lab

1) Create the topic

2) Produce messages to the topic (Python producer)

3) Write the Python consumer

4) Real-world considerations