apache-kafka

Using kafka-python: from kafka import KafkaConsumer if __name__ == ‘__main__’: topic_name=”example-topic” consumer = KafkaConsumer(topic_name, auto_offset_reset=”earliest”, bootstrap_servers=[‘kafka2:9092′], api_version=(0, 10), consumer_timeout_ms=1000) for msg in consumer: print(msg.value) if consumer is not None: consumer.close() from time import sleep from kafka import KafkaProducer # publish messages on topic def publish_message(producer_instance, topic_name, key, value): try: key_bytes = bytes(key, encoding=’utf-8’) value_bytes = … Read more