> ## Documentation Index > Fetch the complete documentation index at: https://notes.kodekloud.com/llms.txt > Use this file to discover all available pages before exploring further. # Demo Kafka Consumers > Guide to creating Kafka topics and writing Python producers and consumers with kafka-python, covering setup, message production, consumption, partitions, offsets, and production considerations 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: ```bash theme={null} 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: ```bash theme={null} cd ~/kafka/bin ./kafka-topics.sh --create --topic multi-partition-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1 ``` Expected successful output: ```bash theme={null} 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`: ```bash theme={null} 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: ```python theme={null} # 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: ```bash theme={null} python3 kafka_producer_example.py ``` You should see logs showing connections and message delivery, for example: ```text theme={null} 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: ```python theme={null} # 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: ```bash theme={null} python3 kafka_consumer.py ``` Expected output (example): ```text theme={null} 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/](https://kafka.apache.org/documentation/) * kafka-python (PyPI): [https://pypi.org/project/kafka-python/](https://pypi.org/project/kafka-python/) * Kafka consumer groups and partitioning: [https://kafka.apache.org/documentation/#consumerapi](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](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.