From 5682ff63534bb2579a6835d40afcad170f6bdd7c Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sat, 5 Dec 2015 04:05:48 -0800 Subject: Producer.stop() now blocks until async thread completes (drop confusing timeout arg) --- kafka/producer/base.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) (limited to 'kafka') diff --git a/kafka/producer/base.py b/kafka/producer/base.py index f2c7cfe..1ba4f5b 100644 --- a/kafka/producer/base.py +++ b/kafka/producer/base.py @@ -415,17 +415,22 @@ class Producer(object): raise return resp - def stop(self, timeout=1): + def stop(self): """ - Stop the producer. Optionally wait for the specified timeout before - forcefully cleaning up. + Stop the producer (async mode). Blocks until async thread completes. """ + if not self.async: + log.warning("producer.stop() called, but producer is not async") + return + + if self.stopped: + log.warning("producer.stop() called, but producer is already stopped") + return + if self.async: self.queue.put((STOP_ASYNC_PRODUCER, None, None)) - self.thread.join(timeout) - - if self.thread.is_alive(): - self.thread_stop_event.set() + self.thread_stop_event.set() + self.thread.join() if hasattr(self, '_cleanup_func'): # Remove cleanup handler now that we've stopped -- cgit v1.2.1