Make external API consistently support python3 strings for topic.

1 files changed, 6 insertions, 0 deletions

diff --git a/kafka/producer/base.py b/kafka/producer/base.py
index 6a5a94e..00c4d46 100644
--- a/kafka/producer/base.py
+++ b/kafka/producer/base.py
@@ -17,6 +17,7 @@ from kafka.common import (
     ProduceRequest, TopicAndPartition, UnsupportedCodecError
 )
 from kafka.protocol import CODEC_NONE, ALL_CODECS, create_message_set
+from kafka.util import kafka_bytestring
 
 log = logging.getLogger("kafka")
 
@@ -170,6 +171,7 @@ class Producer(object):
 
         All messages produced via this method will set the message 'key' to Null
         """
+        topic = kafka_bytestring(topic)
         return self._send_messages(topic, partition, *msg)
 
     def _send_messages(self, topic, partition, *msg, **kwargs):
@@ -183,6 +185,10 @@ class Producer(object):
         if any(not isinstance(m, six.binary_type) for m in msg):
             raise TypeError("all produce message payloads must be type bytes")
 
+        # Raise TypeError if topic is not encoded as bytes
+        if not isinstance(topic, six.binary_type):
+            raise TypeError("the topic must be type bytes")
+
         # Raise TypeError if the key is not encoded as bytes
         if key is not None and not isinstance(key, six.binary_type):
             raise TypeError("the key must be type bytes")