Kafka, Kafka Consumer, Kafka Producer, Confluent Kafka, Streaming.
props: Dict): self.consumer = Consumer(**props) def consume_from_topics(self, topics: List[str]): self.consumer.subscribe(topics=topics) while True: try: msg = self.consumer.poll(1.0) if msg is None: continue else: print("Key:{}, Value:{}".format(msg.key(), msg.value())) except KeyboardInterrupt: