From 3bf75c53bff0038c68eadf6664061b7adb18f64b Mon Sep 17 00:00:00 2001
From: Zack Dever <zdever@pandora.com>
Date: Mon, 14 Mar 2016 16:03:06 -0700
Subject: [PATCH 1/3] only make a metadata event if we don't know the
 partitions

---
 kafka/producer/kafka.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 0286f8b..2a16fd8 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -443,12 +443,15 @@ class KafkaProducer(object):
         self._sender.add_topic(topic)
         begin = time.time()
         elapsed = 0.0
-        metadata_event = threading.Event()
+        metadata_event = None
         while True:
             partitions = self._metadata.partitions_for_topic(topic)
             if partitions is not None:
                 return partitions
 
+            if not metadata_event:
+                metadata_event = threading.Event()
+
             log.debug("Requesting metadata update for topic %s", topic)
 
             metadata_event.clear()

From 8847dbbb0d5fcb504d8c4d738246daf9004e2b19 Mon Sep 17 00:00:00 2001
From: Zack Dever <zdever@pandora.com>
Date: Mon, 14 Mar 2016 16:14:48 -0700
Subject: [PATCH 2/3] make Sender._topics_to_add a set instead of a list

---
 kafka/producer/sender.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py
index 0e6d6cd..bba135b 100644
--- a/kafka/producer/sender.py
+++ b/kafka/producer/sender.py
@@ -45,7 +45,7 @@ class Sender(threading.Thread):
         self._metadata = client.cluster
         self._running = True
         self._force_close = False
-        self._topics_to_add = []
+        self._topics_to_add = set()
 
     def run(self):
         """The main run loop for the sender thread."""
@@ -158,8 +158,9 @@ class Sender(threading.Thread):
         self.initiate_close()
 
     def add_topic(self, topic):
-        self._topics_to_add.append(topic)
-        self.wakeup()
+        if topic not in self._topics_to_add:
+            self._topics_to_add.add(topic)
+            self.wakeup()
 
     def _failed_produce(self, batches, node_id, error):
         log.debug("Error sending produce request to node %d: %s", node_id, error) # trace

From 1b1bd55d03ee56da4b5663aa0dfbfbd8bbc1b73c Mon Sep 17 00:00:00 2001
From: Zack Dever <zdever@pandora.com>
Date: Mon, 14 Mar 2016 16:29:13 -0700
Subject: [PATCH 3/3] remove unused import. needed a change to kick travis.

---
 kafka/producer/sender.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py
index bba135b..9a86a16 100644
--- a/kafka/producer/sender.py
+++ b/kafka/producer/sender.py
@@ -4,7 +4,6 @@ import collections
 import copy
 import logging
 import threading
-import time
 
 import six