mreschke / Simple Kafka Producer/Consumer

import time import json from kafka import KafkaProducer # --- Configuration --- KAFKA_BROKER = ['localhost:29001', 'localhost:29002', 'localhost:29003'] TOPIC_NAME = 'my-test-topic' MESSAGES_TO_SEND = 100 def create_producer(): """Creates and returns a KafkaProducer instance.""" try: producer = KafkaProducer( bootstrap_servers=KAFKA_BROKER, # Serialize messages as JSON and then encode to UTF-8 bytes value_serializer=lambda v: json.dumps(v).encode('utf-8'), # Key serializer key_serializer=str.encode, # Optional: Add acks='all' for higher durability guarantees acks='all', # Optional: Add retries for transient errors retries=5 ) print("Kafka Producer created successfully.") return producer except Exception as e: print(f"Error creating Kafka Producer: {e}") return None def send_message(producer, topic, message_data): """Sends a single message to the specified topic.""" try: # Generate random integer between 1000 and 2000 random_id = int(time.time() * 1000) % 1000 + 1000 # The key can be None, or a specific value (e.g., to ensure messages with the same key go to the same partition) # For this basic example, we'll use None as the key. key = None key = f'client_{random_id}' #key = '21300' #key = f'client-21300' future = producer.send(topic, key=key, value=message_data) # Block for 'synchronous' sends; wait for ack from broker # record_metadata = future.get(timeout=10) # seconds # print(f"Message sent to topic '{record_metadata.topic}' partition {record_metadata.partition} offset {record_metadata.offset}") print(f"Message sent: {message_data}") except Exception as e: print(f"Error sending message: {e}") if __name__ == "__main__": producer = create_producer() if producer: try: for i in range(MESSAGES_TO_SEND): message = { 'id': i, 'text': f'Hello from Python Kafka Producer! Message #{i}', 'timestamp': time.time() } send_message(producer, TOPIC_NAME, message) #time.sleep(0.1) # Wait a bit between messages # Ensure all buffered messages are sent before exiting producer.flush() #time.sleep(1) print(f"All {MESSAGES_TO_SEND} messages sent and flushed.") #time. sleep(1) except KeyboardInterrupt: print("Producer interrupted by user.") finally: if producer: producer.close() print("Kafka Producer closed.")