Wakeup socket optimizationsno_sleep_til_brooklyn

2 files changed, 7 insertions, 2 deletions

diff --git a/kafka/client_async.py b/kafka/client_async.py
index 25952be..63b63d1 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -726,7 +726,7 @@ class KafkaClient(object):
     def _clear_wake_fd(self):
         while True:
             try:
-                self._wake_r.recv(1)
+                self._wake_r.recv(1024)
             except:
                 break
 
diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py
index f0f77ee..958e165 100644
--- a/kafka/producer/sender.py
+++ b/kafka/producer/sender.py
@@ -163,7 +163,12 @@ class Sender(threading.Thread):
         self.initiate_close()
 
     def add_topic(self, topic):
-        if topic not in self._topics_to_add:
+        # This is generally called from a separate thread
+        # so this needs to be a thread-safe operation
+        # we assume that checking set membership across threads
+        # is ok where self._client._topics should never
+        # remove topics for a producer instance, only add them.
+        if topic not in self._client._topics:
             self._topics_to_add.add(topic)
             self.wakeup()