Handle New Topic Creation

Adds ensure_topic_exists to KafkaClient, redirects test case to use
that.  Fixes #113 and fixes #150.

1 files changed, 9 insertions, 0 deletions

diff --git a/test/test_producer_integration.py b/test/test_producer_integration.py
index c69e117..0718cb3 100644
--- a/test/test_producer_integration.py
+++ b/test/test_producer_integration.py
@@ -143,6 +143,15 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
         producer.stop()
 
     @kafka_versions("all")
+    def test_produce__new_topic_fails_with_reasonable_error(self):
+        new_topic = 'new_topic_{}'.format(str(uuid.uuid4()))
+        producer = SimpleProducer(self.client)
+
+        # At first it doesn't exist
+        with self.assertRaises(UnknownTopicOrPartitionError):
+            resp = producer.send_messages(new_topic, self.msg("one"))
+
+    @kafka_versions("all")
     def test_producer_random_order(self):
         producer = SimpleProducer(self.client, random_start = True)
         resp1 = producer.send_messages(self.topic, self.msg("one"), self.msg("two"))