diff options
| -rw-r--r-- | README.md | 4 | ||||
| -rw-r--r-- | kafka/producer.py | 6 |
2 files changed, 9 insertions, 1 deletions
@@ -51,6 +51,8 @@ producer.send_messages("my-topic", "this method", "is variadic") producer.send_messages("my-topic", u'你怎么样?'.encode('utf-8')) # To send messages asynchronously +# WARNING: current implementation does not guarantee message delivery on failure! +# messages can get dropped! Use at your own risk! Or help us improve with a PR! producer = SimpleProducer(kafka, async=True) producer.send_messages("my-topic", "async message") @@ -63,7 +65,7 @@ producer = SimpleProducer(kafka, async=False, req_acks=SimpleProducer.ACK_AFTER_LOCAL_WRITE, ack_timeout=2000) -response = producer.send_messages("my-topic", "async message") +response = producer.send_messages("my-topic", "another message") if response: print(response[0].error) diff --git a/kafka/producer.py b/kafka/producer.py index 800e677..8a6bff0 100644 --- a/kafka/producer.py +++ b/kafka/producer.py @@ -87,6 +87,9 @@ class Producer(object): client - The Kafka client instance to use async - If set to true, the messages are sent asynchronously via another thread (process). We will not wait for a response to these + WARNING!!! current implementation of async producer does not + guarantee message delivery. Use at your own risk! Or help us + improve with a PR! req_acks - A value indicating the acknowledgements that the server must receive before responding to the request ack_timeout - Value (in milliseconds) indicating a timeout for waiting @@ -131,6 +134,9 @@ class Producer(object): self.codec = codec if self.async: + log.warning("async producer does not guarantee message delivery!") + log.warning("Current implementation does not retry Failed messages") + log.warning("Use at your own risk! (or help improve with a PR!)") self.queue = Queue() # Messages are sent through this queue self.proc = Process(target=_send_upstream, args=(self.queue, |