This tutorial provides a complete Python script for interacting with Apache Kafka as both a producer and a consumer using the confluent-kafka library. It c
This tutorial provides a complete Python script for interacting with Apache Kafka as both a producer and a consumer using the confluent-kafka library. It covers configuration, usage, error handling, and ideas for extension, making it suitable for production environments.
In modern distributed systems, applications often need to communicate asynchronously, process real-time data streams, or reliably ingest large volumes of events. Traditional request-response patterns or direct database writes can become bottlenecks or single points of failure. Apache Kafka addresses these challenges by acting as a high-throughput, fault-tolerant, and scalable publish-subscribe messaging system. This script provides a fundamental building block for any Python application that needs to send data into Kafka (as a producer) or read data from Kafka (as a consumer), enabling real-time analytics, event sourcing, log aggregation, and microservices communication.
The real-world problem solved here is providing a robust and easy-to-understand template for Python developers to integrate their applications with Kafka. It simplifies the setup of basic producer-consumer patterns, allowing developers to focus on their application's business logic rather than the intricacies of Kafka client configuration and interaction.
This script operates in two distinct modes: producer and consumer, determined by a command-line argument. Both modes leverage the confluent-kafka client library, which provides a high-performance, C-based Kafka client wrapper for Python.
As a producer, the script generates sample JSON messages containing an ID and a timestamp. These messages are serialized to bytes and sent to a specified Kafka topic. The producer uses an asynchronous approach, buffering messages and sending them in batches to optimize throughput. A delivery_report callback is used to confirm successful message delivery or log any errors, ensuring that the script has visibility into the publishing process.
As a consumer, the script subscribes to a specified Kafka topic and continuously polls for new messages. Upon receiving a message, it decodes the message value (assuming it's a JSON string), prints it, and then explicitly commits its offset. Manual offset committing is chosen over automatic committing for greater control and to ensure that messages are only considered processed after they have been successfully handled by the application logic, providing "at least once" delivery semantics. Configuration is handled via environment variables for Kafka broker addresses and command-line arguments for runtime choices like topic and consumer group, making it flexible and easy to deploy across different environments.
#!/usr/bin/env python3
"""
Kafka Producer and Consumer with confluent-kafka
This script demonstrates how to produce and consume messages from Apache Kafka
using the confluent-kafka Python library. It supports both producer and consumer
modes, allowing users to send sample JSON data to a topic or read data from it.
Configuration for Kafka brokers is primarily handled via environment variables.
"""
import os
import sys
import argparse
import json
import time
import logging
from typing import Dict, Any, Optional
from confluent_kafka import Producer, Consumer, KafkaException, Message
# --- Configuration Section ---
# Set up basic logging
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# Kafka broker configuration via environment variable
KAFKA_BOOTSTRAP_SERVERS: str = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
# Default topic name
DEFAULT_TOPIC: str = os.environ.get("DEFAULT_TOPIC", "my_events")
# Default consumer group ID
DEFAULT_GROUP_ID: str = os.environ.get("DEFAULT_GROUP_ID", "my_app_group")
# --- Producer Callbacks and Logic ---
def delivery_report(err: Optional[KafkaException], msg: Message) -> None:
"""
Callback function called once for each message produced to indicate delivery status.
Triggered by poll() or flush().
"""
if err is not None:
logger.error(f"Message delivery failed for key {msg.key()}: {err}")
else:
logger.info(f"Message delivered to topic '{msg.topic()}' "
f"[partition {msg.partition()}] @ offset {msg.offset()} "
f"with key {msg.key().decode('utf-8') if msg.key() else 'N/A'}")
def run_producer(bootstrap_servers: str, topic: str, num_messages: int = 10) -> None:
"""
Runs the Kafka producer to send messages to the specified topic.
"""
logger.info(f"Starting Kafka producer for topic: {topic}")
logger.info(f"Kafka bootstrap servers: {bootstrap_servers}")
# Producer configuration
producer_conf: Dict[str, Any] = {
'bootstrap.servers': bootstrap_servers,
'client.id': 'python-kafka-producer',
'acks': 'all', # Ensure all in-sync replicas have received the message
'retries': 3, # Number of retries before giving up
'linger.ms': 50, # Time to wait before sending messages in batches
'enable.idempotence': True # Guarantees exactly-once delivery within a single session
}
producer: Producer = Producer(producer_conf)
try:
for i in range(num_messages):
message_id: int = i + 1
current_time: float = time.time()
# Create a sample JSON message
message_payload: Dict[str, Any] = {
"id": message_id,
"timestamp": current_time,
"event_type": "user_action",
"data": f"user_clicked_item_{message_id}"
}
# Encode message key and value
key: bytes = str(message_id).encode('utf-8')
value: bytes = json.dumps(message_payload).encode('utf-8')
# Produce message
# The produce() method is asynchronous. It buffers the message and
# returns immediately. The delivery_report callback will be invoked
# once the message is successfully delivered or fails.
try:
producer.produce(topic, key=key, value=value, callback=delivery_report)
# Poll for events to trigger callbacks (e.g., delivery reports)
# This is important to ensure callbacks are processed.
producer.poll(0)
logger.debug(f"Queued message {message_id} for topic {topic}")
except BufferError:
logger.warning(f"Local producer queue is full ({len(producer)} messages awaiting delivery): "
f"waiting for messages to be delivered...")
# If the buffer is full, flush it to clear space and then retry
producer.flush(timeout=10) # Flush with a timeout
producer.produce(topic, key=key, value=value, callback=delivery_report)
producer.poll(0)
time.sleep(0.1) # Simulate some work or delay
except KeyboardInterrupt:
logger.info("Producer interrupted by user.")
except KafkaException as e:
logger.error(f"Kafka producer error: {e}")
finally:
# It's essential to call flush() at the end to ensure all buffered
# messages are sent before the producer exits.
remaining_messages: int = producer.flush(timeout=10)
if remaining_messages > 0:
logger.warning(f"Producer: {remaining_messages} messages still in queue after flush.")
logger.info("Kafka producer finished.")
# --- Consumer Logic ---
def run_consumer(bootstrap_servers: str, topic: str, group_id: str) -> None:
"""
Runs the Kafka consumer to read messages from the specified topic.
"""
logger.info(f"Starting Kafka consumer for topic: {topic}, group: {group_id}")
logger.info(f"Kafka bootstrap servers: {bootstrap_servers}")
# Consumer configuration
consumer_conf: Dict[str, Any] = {
'bootstrap.servers': bootstrap_servers,
'group.id': group_id,
'auto.offset.reset': 'earliest', # Start consuming from the beginning if no offset is found
'enable.auto.commit': False, # Disable auto commit for explicit control
'client.id': 'python-kafka-consumer',
'session.timeout.ms': 10000, # Session timeout for group coordination
'max.poll.interval.ms': 300000 # Max time between polls before consumer is considered dead
}
consumer: Consumer = Consumer(consumer_conf)
try:
# Subscribe to the topic(s)
consumer.subscribe([topic])
logger.info(f"Consumer subscribed to topic: {topic}")
while True:
# Poll for messages with a timeout. If no message is available within
# the timeout, msg will be None.
msg: Optional[Message] = consumer.poll(timeout=1.0)
if msg is None:
logger.debug("Waiting for messages...")
time.sleep(0.5) # Small delay to avoid busy-waiting
continue
if msg.error():
if msg.error().is_fatal():
logger.error(f"Fatal consumer error: {msg.error()}")
raise KafkaException(msg.error())
else:
logger.warning(f"Consumer error: {msg.error()}")
continue # Non-fatal error, continue polling
else:
# Proper message received
try:
key: Optional[bytes] = msg.key()
value: Optional[bytes] = msg.value()
if value is None:
logger.warning(f"Received message with null value from topic {msg.topic()} "
f"[partition {msg.partition()}] @ offset {msg.offset()}")
continue
decoded_key: str = key.decode('utf-8') if key else "N/A"
decoded_value: Dict[str, Any] = json.loads(value.decode('utf-8'))
logger.info(f"Received message: "
f"Topic: {msg.topic()}, Partition: {msg.partition()}, Offset: {msg.offset()}, "
f"Key: {decoded_key}, Value: {decoded_value}")
# Explicitly commit the offset after successful processing
# This ensures that if the consumer crashes, it will restart
# from the last successfully processed message.
consumer.commit(message=msg)
logger.debug(f"Committed offset {msg.offset()} for partition {msg.partition()}")
except json.JSONDecodeError:
logger.error(f"Failed to decode JSON from message value: {msg.value()}")
except UnicodeDecodeError:
logger.error(f"Failed to decode message value as UTF-8: {msg.value()}")
except Exception as e:
logger.error(f"Error processing message: {e}", exc_info=True)
except KeyboardInterrupt:
logger.info("Consumer interrupted by user.")
except KafkaException as e:
logger.error(f"Kafka consumer error: {e}")
finally:
# Close the consumer to commit final offsets and leave the consumer group gracefully.
logger.info("Closing Kafka consumer.")
consumer.close()
# --- Main Entry Point ---
def main() -> None:
"""
Parses command-line arguments and runs the Kafka producer or consumer.
"""
parser = argparse.ArgumentParser(
description="Kafka Producer and Consumer with confluent-kafka",
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument(
"--mode",
type=str,
choices=["producer", "consumer"],
required=True,
help="Operation mode: 'producer' to send messages, 'consumer' to receive messages."
)
parser.add_argument(
"--topic",
type=str,
default=DEFAULT_TOPIC,
help="Kafka topic to produce to or consume from."
)
parser.add_argument(
"--group",
type=str,
default=DEFAULT_GROUP_ID,
help="Consumer group ID (only applicable for consumer mode)."
)
parser.add_argument(
"--num-messages",
type=int,
default=10,
help="Number of messages to produce (only applicable for producer mode)."
)
parser.add_argument(
"--bootstrap-servers",
type=str,
default=KAFKA_BOOTSTRAP_SERVERS,
help="Comma-separated list of Kafka bootstrap servers (host:port)."
)
args = parser.parse_args()
if args.mode == "producer":
run_producer(args.bootstrap_servers, args.topic, args.num_messages)
elif args.mode == "consumer":
run_consumer(args.bootstrap_servers, args.topic, args.group)
if __name__ == "__main__":
main()
The script primarily uses environment variables for sensitive or environment-specific configurations and command-line arguments for runtime choices.
KAFKA_BOOTSTRAP_SERVERS:
host1:port1,host2:port2). The client will use this list to connect to the Kafka cluster.localhost:9092 (suitable for a local Kafka setup or Docker Compose).export KAFKA_BOOTSTRAP_SERVERS="kafka1:9092,kafka2:9092"DEFAULT_TOPIC:
--topic command-line argument.my_eventsexport DEFAULT_TOPIC="production_logs"DEFAULT_GROUP_ID:
--group command-line argument. Consumer groups are essential for load balancing and fault tolerance among multiple consumer instances.my_app_groupexport DEFAULT_GROUP_ID="log_processor_group"LOG_LEVEL:
INFOexport LOG_LEVEL="DEBUG"Command-line Arguments:
--mode: Required. Specifies whether to run as a producer or consumer.--topic: Overrides DEFAULT_TOPIC.--group: Overrides DEFAULT_GROUP_ID (consumer mode only).--num-messages: Number of messages to produce (producer mode only, default 10).--bootstrap-servers: Overrides KAFKA_BOOTSTRAP_SERVERS directly (useful for testing without setting env vars).First, ensure you have confluent-kafka installed:
pip install confluent-kafka
And ideally, a Kafka broker running (e.g., via Docker):
docker run -p 9092:9092 -e KAFKA_BROKER_ID=1 \
-e KAFKA_ZOOKEEPER_CONNECT=localhost:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
-d confluentinc/cp-kafka:7.6.0
# You might also need a Zookeeper, e.g., docker run -p 2181:2181 -d zookeeper:3.8.0
This command will produce 5 messages to the topic my_events (default) on localhost:9092 (default).
python3 kafka_client.py --mode producer --num-messages 5
Expected Output (Producer):
INFO - __main__ - Starting Kafka producer for topic: my_events
INFO - __main__ - Kafka bootstrap servers: localhost:9092
INFO - __main__ - Message delivered to topic 'my_events' [partition 0] @ offset 0 with key 1
INFO - __main__ - Message delivered to topic 'my_events' [partition 0] @ offset 1 with key 2
INFO - __main__ - Message delivered to topic 'my_events' [partition 0] @ offset 2 with key 3
INFO - __main__ - Message delivered to topic 'my_events' [partition 0] @ offset 3 with key 4
INFO - __main__ - Message delivered to topic 'my_events' [partition 0] @ offset 4 with key 5
INFO - __main__ - Kafka producer finished.
This command will consume messages from the topic my_events (default) using consumer group my_app_group (default). Run this in a separate terminal after or while the producer is running.
python3 kafka_client.py --mode consumer
Expected Output (Consumer):
INFO - __main__ - Starting Kafka consumer for topic: my_events, group: my_app_group
INFO - __main__ - Kafka bootstrap servers: localhost:9092
INFO - __main__ - Consumer subscribed to topic: my_events
INFO - __main__ - Received message: Topic: my_events, Partition: 0, Offset: 0, Key: 1, Value: {'id': 1, 'timestamp': 1678886400.123, 'event_type': 'user_action', 'data': '