Block integration tests until topic gets leader

1 files changed, 11 insertions, 1 deletions

diff --git a/test/testutil.py b/test/testutil.py
index 0bacac4..0ec1cff 100644
--- a/test/testutil.py
+++ b/test/testutil.py
@@ -12,6 +12,7 @@ from six.moves import xrange
 from . import unittest
 
 from kafka import SimpleClient
+from kafka.errors import LeaderNotAvailableError, KafkaTimeoutError
 from kafka.structs import OffsetRequestPayload
 
 __all__ = [
@@ -98,7 +99,16 @@ class KafkaIntegrationTestCase(unittest.TestCase):
         if self.create_client:
             self.client = SimpleClient('%s:%d' % (self.server.host, self.server.port))
 
-        self.client.ensure_topic_exists(self.topic)
+        timeout = time.time() + 30
+        while time.time() < timeout:
+            try:
+                self.client.load_metadata_for_topics(self.topic, ignore_leadernotavailable=False)
+                if self.client.has_metadata_for_topic(topic):
+                    break
+            except LeaderNotAvailableError:
+                time.sleep(1)
+        else:
+            raise KafkaTimeoutError('Timeout loading topic metadata!')
 
         self._messages = {}