Kafka, Kafka Consumer, Kafka Producer, Confluent Kafka, Streaming.
BufferError as bfer: self.producer.poll(0.1) # Polls the producer for events and calls the corresponding callbacks (if registered). except Exception as e: print(f"Exception while producing message - {message}: {e}") except KeyboardInterrupt: break self.producer.flush() # Wait for all messages in