Digging into a new universe of data streaming after IBM’s recent accquisition of Confluent!
Using two basic code samples I found on the net, one as a “producer” and one as a “consumer” to explain the concepts of Kafka, I asked Bob to build a simple comprehensive application.
- Simple Producer
CODE
from confluent_kafka import Producer
import json
# Configuration for connecting to the Kafka cluster
config = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(config)
def delivery_report(err, msg):
if err is not None:
print(f"Message delivery failed: {err}")
else:
print(f"Order sent to {msg.topic()} [{msg.partition()}]")
# Simulate an order
order_data = {
"order_id": 1001,
"user": "jane_doe",
"total": 59.99,
"items": ["Wireless Mouse", "Keyboard"]
}
# Trigger the send (Asynchronous)
producer.produce(
'orders',
key="1001",
value=json.dumps(order_data),
callback=delivery_report
)
producer.flush() # Wait for any outstanding messages to be delivered
- Simple Consumer
CODE
from confluent_kafka import Consumer
config = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'shipping-group', # Helps Kafka track which orders this group has seen
'auto.offset.reset': 'earliest'
}
consumer = Consumer(config)
consumer.subscribe(['orders'])
try:
while True:
msg = consumer.poll(1.0) # Check for new messages every 1 second
if msg is None: continue
if msg.error():
print(f"Consumer error: {msg.error()}")
continue
print(f"Received Order: {msg.value().decode('utf-8')}")
print("Action: Initiating packaging and shipping...")
finally:
consumer.close()
After reading my conceptual samples, Bob provided the application structure which follows.
The Producer: Capturing the Event
The Producer (found in src/producer.py) acts as the storefront. When an order is placed, it creates a JSON-serializable dictionary containing the order ID, user details, and items. It uses the confluent-kafka library to "produce" this message to a topic called orders, using the order_id as a key to ensure all updates for a specific order land in the same partition. A delivery callback function is utilized to provide immediate feedback on whether the message successfully reached the Kafka broker or if a retry is necessary.
The Consumer logic (found in src/consumer.py) demonstrates the power of Consumer Groups. Multiple independent services—Shipping, Email, and Analytics—all subscribe to the same orders topic simultaneously. Because each service belongs to its own group (e.g., shipping-group), Kafka tracks their progress (offsets) individually. This means the Shipping department can process messages at its own pace without affecting the speed of the Email service or the Analytics dashboard.
CODE
"""
Kafka Consumer - E-commerce Order Processing System
Simulates different departments (Shipping, Email, Analytics) consuming order events
"""
from confluent_kafka import Consumer, KafkaError
import json
import sys
from datetime import datetime
def create_consumer(group_id, bootstrap_servers='localhost:9092'):
"""
Create and configure a Kafka consumer
Args:
group_id: Consumer group identifier
bootstrap_servers: Kafka broker addresses
Returns:
Configured Consumer instance
"""
config = {
'bootstrap.servers': bootstrap_servers,
'group.id': group_id,
'auto.offset.reset': 'earliest', # Start from beginning if no offset exists
'enable.auto.commit': True,
'auto.commit.interval.ms': 1000,
'session.timeout.ms': 6000,
'client.id': f'{group_id}-client'
}
return Consumer(config)
def process_shipping(order):
"""Process order for shipping department"""
print(f"\n📦 SHIPPING DEPARTMENT")
print(f" Order ID: {order['order_id']}")
print(f" Customer: {order['user']}")
print(f" Items to pack: {', '.join(order['items'])}")
print(f" ✅ Initiating packaging and shipping process...")
def process_email(order):
"""Process order for email service"""
print(f"\n📧 EMAIL SERVICE")
print(f" Order ID: {order['order_id']}")
print(f" Recipient: {order['user']}")
print(f" Total: ${order['total']}")
print(f" ✅ Sending order confirmation email...")
def process_analytics(order):
"""Process order for analytics dashboard"""
print(f"\n📊 ANALYTICS DASHBOARD")
print(f" Order ID: {order['order_id']}")
print(f" Revenue: ${order['total']}")
print(f" Items count: {len(order['items'])}")
print(f" Timestamp: {order['timestamp']}")
print(f" ✅ Updating live sales dashboard...")
# Department processors mapping
PROCESSORS = {
'shipping-group': process_shipping,
'email-group': process_email,
'analytics-group': process_analytics
}
def consume_orders(group_id, topic='orders'):
"""
Consume orders from Kafka topic
Args:
group_id: Consumer group (shipping-group, email-group, analytics-group)
topic: Kafka topic to subscribe to
"""
consumer = create_consumer(group_id)
consumer.subscribe([topic])
processor = PROCESSORS.get(group_id, process_shipping)
department = group_id.replace('-group', '').upper()
print(f"🚀 Starting {department} Consumer")
print(f"👥 Consumer Group: {group_id}")
print(f"📋 Subscribed to topic: {topic}")
print(f"🔗 Connected to Kafka at: localhost:9092")
print("-" * 70)
print("⏳ Waiting for messages... (Press Ctrl+C to stop)\n")
try:
message_count = 0
while True:
# Poll for messages (timeout in seconds)
msg = consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
print(f"📍 Reached end of partition {msg.partition()}")
else:
print(f"❌ Consumer error: {msg.error()}")
continue
# Process the message
try:
message_count += 1
order_data = json.loads(msg.value().decode('utf-8'))
print(f"\n{'='*70}")
print(f"📨 Message #{message_count} received from partition {msg.partition()} at offset {msg.offset()}")
# Call the appropriate processor
processor(order_data)
print(f"{'='*70}")
except json.JSONDecodeError as e:
print(f"❌ Failed to decode message: {e}")
except Exception as e:
print(f"❌ Error processing message: {e}")
except KeyboardInterrupt:
print(f"\n\n⚠️ Consumer interrupted by user")
print(f"📊 Total messages processed: {message_count}")
finally:
# Close the consumer to commit final offsets
print("🔒 Closing consumer...")
consumer.close()
print("✅ Consumer closed successfully")
if __name__ == "__main__":
# Parse command line arguments
if len(sys.argv) < 2:
print("Usage: python consumer.py <group_id> [topic]")
print("\nAvailable consumer groups:")
print(" - shipping-group : Processes orders for shipping")
print(" - email-group : Sends confirmation emails")
print(" - analytics-group : Updates analytics dashboard")
print("\nExample: python consumer.py shipping-group")
sys.exit(1)
group_id = sys.argv[1]
topic = sys.argv[2] if len(sys.argv) > 2 else 'orders'
if group_id not in PROCESSORS:
print(f"⚠️ Warning: Unknown group_id '{group_id}'. Using default processor.")
try:
consume_orders(group_id, topic)
except Exception as e:
print(f"❌ Fatal error: {e}")
sys.exit(1)
# Made with Bob
The Cluster: Ensuring Reliability
Vollständiger Original-Bericht
Ausführliche Details, Code-Beispiele & Hersteller-Stellungnahme auf dev.to.
SOCIAL SHARE CARD GENERATOR