pykafka

import os import time from pykafka import KafkaClient from pykafka.common import OffsetType KAFKA_HOSTS = os.environ.get('KAFKA_HOSTS', 'localhost:9092') TOPIC_NAME = os.environ.get('KAFKA_TOPIC', 'test-topic').encode('utf-8') CONSUMER_GROUP = os.environ.get('KAFKA_CONSUMER_GROUP', 'my-consumer-group').encode('utf-8') # 1. Connect to Kafka client = KafkaClient(hosts=KAFKA_HOSTS) topic = client.topics[TOPIC_NAME] # 2. Produce a message with topic.get_producer() as producer: print(f"Producing message to {TOPIC_NAME.decode()}...") for i in range(5): message_value = f"test message {i}".encode('utf-8') producer.produce(message_value) print(f"Sent: {message_value.decode()}") print("Finished producing messages.") # Give Kafka a moment to process time.sleep(2) # 3. Consume messages print(f"\nConsuming messages from {TOPIC_NAME.decode()} (group: {CONSUMER_GROUP.decode()})...") consumer = topic.get_simple_consumer( consumer_group=CONSUMER_GROUP, auto_commit_enable=True, auto_offset_reset=OffsetType.EARLIEST, # Start from the beginning if no offset is committed reset_offset_on_start=True # Force reset on start, useful for quickstart ) for i, message in enumerate(consumer): if message is not None: print(f"Received: Offset={message.offset}, Value={message.value.decode()}") if i >= 4: # Consume the 5 messages we sent break consumer.stop() print("Finished consuming messages.")