Add KafkaTimeoutError (used by client.ensure_topic_exists) and add a test

1 files changed, 11 insertions, 0 deletions

diff --git a/test/test_client_integration.py b/test/test_client_integration.py
index 49c4b57..98f2473 100644
--- a/test/test_client_integration.py
+++ b/test/test_client_integration.py
@@ -49,6 +49,17 @@ class TestKafkaClientIntegration(KafkaIntegrationTestCase):
         messages = list(fetch_resp.messages)
         self.assertEquals(len(messages), 0)
 
+    @kafka_versions("all")
+    def test_ensure_topic_exists(self):
+
+        # assume that self.topic was created by setUp
+        # if so, this should succeed
+        self.client.ensure_topic_exists(self.topic, timeout=1)
+
+        # ensure_topic_exists should fail with KafkaTimeoutError
+        with self.assertRaises(KafkaTimeoutError):
+            self.client.ensure_topic_exists("this_topic_doesnt_exist", timeout=0)
+
     ####################
     #   Offset Tests   #
     ####################