kafka-python

import os from kafka import KafkaProducer, KafkaConsumer import json import time # Configuration KAFKA_BOOTSTRAP_SERVERS = os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092') KAFKA_TOPIC = os.environ.get('KAFKA_TOPIC', 'my_test_topic') # --- Producer Example --- def produce_messages(): producer = KafkaProducer( bootstrap_servers=[KAFKA_BOOTSTRAP_SERVERS], value_serializer=lambda v: json.dumps(v).encode('utf-8') ) print(f"Producing messages to topic: {KAFKA_TOPIC}") for i in range(5): message = {'number': i, 'timestamp': time.time()} future = producer.send(KAFKA_TOPIC, message) try: record_metadata = future.get(timeout=10) print(f"Sent: {message} to partition {record_metadata.partition} offset {record_metadata.offset}") except Exception as e: print(f"Error sending message: {e}") producer.flush() # Ensure all messages are sent producer.close() print("Producer finished.") # --- Consumer Example --- def consume_messages(): consumer = KafkaConsumer( KAFKA_TOPIC, bootstrap_servers=[KAFKA_BOOTSTRAP_SERVERS], auto_offset_reset='earliest', # Start reading from the beginning of the topic if no committed offset enable_auto_commit=True, # Auto-commit offsets periodically group_id='my_python_group', # Required for auto-assignment and offset commits value_deserializer=lambda x: json.loads(x.decode('utf-8')) ) print(f"Consuming messages from topic: {KAFKA_TOPIC}") try: for message in consumer: # message value and key are deserialized print(f"Received: topic={message.topic}, partition={message.partition}, offset={message.offset}, key={message.key}, value={message.value}") except KeyboardInterrupt: print("Consumer interrupted.") finally: consumer.close() print("Consumer closed.") if __name__ == '__main__': # Make sure a Kafka broker is running at KAFKA_BOOTSTRAP_SERVERS (e.g., localhost:9092) # And the topic 'my_test_topic' exists (or let Kafka create it if configured) import threading # Run producer in a separate thread producer_thread = threading.Thread(target=produce_messages) producer_thread.start() # Give producer a moment to start and send some messages time.sleep(2) # Run consumer in the main thread (or another thread, but be mindful of thread-safety) consume_messages() producer_thread.join()