From b687b4c5d4788d64efe9b7bcfb776e57d6fbcc8e Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@rd.io>
Date: Sat, 5 Dec 2015 03:36:12 -0800
Subject: Use producer.stop() to flush messages in async producer test

---
 test/test_producer_integration.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/test/test_producer_integration.py b/test/test_producer_integration.py
index 46b6851..c99ed63 100644
--- a/test/test_producer_integration.py
+++ b/test/test_producer_integration.py
@@ -204,13 +204,11 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
         resp = producer.send_messages(self.topic, self.msg("one"))
         self.assertEqual(len(resp), 0)
 
-        # wait for the server to report a new highwatermark
-        while self.current_offset(self.topic, partition) == start_offset:
-          time.sleep(0.1)
+        # flush messages
+        producer.stop()
 
         self.assert_fetch_offset(partition, start_offset, [ self.msg("one") ])
 
-        producer.stop()
 
     @kafka_versions("all")
     def test_batched_simple_producer__triggers_by_message(self):
-- 
cgit v1.2.1


From 2916bb865085ae2b883bc74dc4988b6b7723917d Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@rd.io>
Date: Sat, 5 Dec 2015 03:59:09 -0800
Subject: Revert broken send_produce_request try/except from PR 467
 (resps=None)

---
 kafka/client.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/kafka/client.py b/kafka/client.py
index b3dcf51..9018bb4 100644
--- a/kafka/client.py
+++ b/kafka/client.py
@@ -607,11 +607,7 @@ class KafkaClient(object):
         else:
             decoder = KafkaProtocol.decode_produce_response
 
-        try:
-            resps = self._send_broker_aware_request(payloads, encoder, decoder)
-        except Exception:
-            if fail_on_error:
-                raise
+        resps = self._send_broker_aware_request(payloads, encoder, decoder)
 
         return [resp if not callback else callback(resp) for resp in resps
                 if resp is not None and
-- 
cgit v1.2.1


From 5682ff63534bb2579a6835d40afcad170f6bdd7c Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@rd.io>
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(-)

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