diff options
| author | Dana Powers <dana.powers@rd.io> | 2014-08-13 11:51:48 -0700 |
|---|---|---|
| committer | Dana Powers <dana.powers@rd.io> | 2014-08-14 09:26:16 -0700 |
| commit | 18fdddf37f552cef72bd7714c1aacc8a35a5f190 (patch) | |
| tree | 64a195c9f959a5203246be1c7562beaeb16646d1 | |
| parent | 5a02d6393d573ca8512a7032a35391bc5ac96f6b (diff) | |
| download | kafka-python-18fdddf37f552cef72bd7714c1aacc8a35a5f190.tar.gz | |
Add KafkaTimeoutError (used by client.ensure_topic_exists) and add a test
| -rw-r--r-- | kafka/client.py | 6 | ||||
| -rw-r--r-- | kafka/common.py | 4 | ||||
| -rw-r--r-- | test/test_client_integration.py | 11 |
3 files changed, 18 insertions, 3 deletions
diff --git a/kafka/client.py b/kafka/client.py index 9cb4b48..9474091 100644 --- a/kafka/client.py +++ b/kafka/client.py @@ -8,8 +8,8 @@ import kafka.common from kafka.common import (TopicAndPartition, ConnectionError, FailedPayloadsError, - PartitionUnavailableError, - LeaderUnavailableError, KafkaUnavailableError, + PartitionUnavailableError, LeaderUnavailableError, KafkaUnavailableError, + KafkaTimeoutError, UnknownTopicOrPartitionError, NotLeaderForPartitionError) from kafka.conn import collect_hosts, KafkaConnection, DEFAULT_SOCKET_TIMEOUT_SECONDS @@ -219,7 +219,7 @@ class KafkaClient(object): self.load_metadata_for_topics(topic) while not self.has_metadata_for_topic(topic): if time.time() > start_time + timeout: - raise KafkaTimeoutError("Unable to create topic {}".format(topic)) + raise KafkaTimeoutError("Unable to create topic {0}".format(topic)) self.load_metadata_for_topics(topic) time.sleep(.5) diff --git a/kafka/common.py b/kafka/common.py index 209754b..907e128 100644 --- a/kafka/common.py +++ b/kafka/common.py @@ -135,6 +135,10 @@ class KafkaUnavailableError(KafkaError): pass +class KafkaTimeoutError(KafkaError): + pass + + class LeaderUnavailableError(KafkaError): pass 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 # #################### |