Drop kafka_bytestring

14 files changed, 47 insertions, 81 deletions

diff --git a/kafka/client.py b/kafka/client.py
index cb60d98..ca737c4 100644
--- a/kafka/client.py
+++ b/kafka/client.py
@@ -17,7 +17,6 @@ from kafka.common import (TopicAndPartition, BrokerMetadata, UnknownError,
 
 from kafka.conn import collect_hosts, BrokerConnection, DEFAULT_SOCKET_TIMEOUT_SECONDS
 from kafka.protocol import KafkaProtocol
-from kafka.util import kafka_bytestring
 
 
 log = logging.getLogger(__name__)
@@ -212,7 +211,7 @@ class KafkaClient(object):
                 failed_payloads(broker_payloads)
                 continue
 
-            conn = self._get_conn(broker.host.decode('utf-8'), broker.port)
+            conn = self._get_conn(broker.host, broker.port)
             request = encoder_fn(payloads=broker_payloads)
             # decoder_fn=None signal that the server is expected to not
             # send a response.  This probably only applies to
@@ -305,7 +304,7 @@ class KafkaClient(object):
 
         # Send the request, recv the response
         try:
-            conn = self._get_conn(broker.host.decode('utf-8'), broker.port)
+            conn = self._get_conn(broker.host, broker.port)
             conn.send(requestId, request)
 
         except ConnectionError as e:
@@ -410,14 +409,12 @@ class KafkaClient(object):
         self.topic_partitions.clear()
 
     def has_metadata_for_topic(self, topic):
-        topic = kafka_bytestring(topic)
         return (
           topic in self.topic_partitions
           and len(self.topic_partitions[topic]) > 0
         )
 
     def get_partition_ids_for_topic(self, topic):
-        topic = kafka_bytestring(topic)
         if topic not in self.topic_partitions:
             return []
 
diff --git a/kafka/consumer/base.py b/kafka/consumer/base.py
index 5859d36..4ac8c66 100644
--- a/kafka/consumer/base.py
+++ b/kafka/consumer/base.py
@@ -11,7 +11,7 @@ from kafka.common import (
     UnknownTopicOrPartitionError, check_error, KafkaError
 )
 
-from kafka.util import kafka_bytestring, ReentrantTimer
+from kafka.util import ReentrantTimer
 
 
 log = logging.getLogger('kafka.consumer')
@@ -47,8 +47,8 @@ class Consumer(object):
                  auto_commit_every_t=AUTO_COMMIT_INTERVAL):
 
         self.client = client
-        self.topic = kafka_bytestring(topic)
-        self.group = None if group is None else kafka_bytestring(group)
+        self.topic = topic
+        self.group = group
         self.client.load_metadata_for_topics(topic)
         self.offsets = {}
 
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 4a630ed..5d91469 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -16,7 +16,6 @@ from kafka.common import (
     OffsetOutOfRangeError, RequestTimedOutError, KafkaMessage, ConsumerTimeout,
     FailedPayloadsError, KafkaUnavailableError, KafkaConfigurationError
 )
-from kafka.util import kafka_bytestring
 
 logger = logging.getLogger(__name__)
 
@@ -307,13 +306,13 @@ class KafkaConsumer(object):
 
             # Topic name str -- all partitions
             if isinstance(arg, (six.string_types, six.binary_type)):
-                topic = kafka_bytestring(arg)
+                topic = arg
                 for partition in self._cluster.partitions_for_topic(topic):
                     self._consume_topic_partition(topic, partition)
 
             # (topic, partition [, offset]) tuple
             elif isinstance(arg, tuple):
-                topic = kafka_bytestring(arg[0])
+                topic = arg[0]
                 partition = arg[1]
                 self._consume_topic_partition(topic, partition)
                 if len(arg) == 3:
@@ -326,7 +325,7 @@ class KafkaConsumer(object):
 
                     # key can be string (a topic)
                     if isinstance(key, (six.string_types, six.binary_type)):
-                        topic = kafka_bytestring(key)
+                        topic = key
 
                         # topic: partition
                         if isinstance(value, int):
@@ -344,7 +343,7 @@ class KafkaConsumer(object):
 
                     # (topic, partition): offset
                     elif isinstance(key, tuple):
-                        topic = kafka_bytestring(key[0])
+                        topic = key[0]
                         partition = key[1]
                         self._consume_topic_partition(topic, partition)
                         self._offsets.fetch[(topic, partition)] = value
@@ -463,7 +462,7 @@ class KafkaConsumer(object):
                 self._refresh_metadata_on_error()
                 continue
 
-            topic = kafka_bytestring(resp.topic)
+            topic = resp.topic
             partition = resp.partition
             try:
                 check_error(resp)
@@ -662,7 +661,7 @@ class KafkaConsumer(object):
         if commits:
             logger.info('committing consumer offsets to group %s', self._config['group_id'])
             resps = self._client.send_offset_commit_request(
-                kafka_bytestring(self._config['group_id']), commits,
+                self._config['group_id'], commits,
                 fail_on_error=False
             )
 
@@ -725,7 +724,7 @@ class KafkaConsumer(object):
         logger.info("Consumer fetching stored offsets")
         for topic_partition in self._topics:
             (resp,) = self._client.send_offset_fetch_request(
-                kafka_bytestring(self._config['group_id']),
+                self._config['group_id'],
                 [OffsetFetchRequest(topic_partition[0], topic_partition[1])],
                 fail_on_error=False)
             try:
diff --git a/kafka/consumer/kafka.py b/kafka/consumer/kafka.py
index fa70124..cdf8760 100644
--- a/kafka/consumer/kafka.py
+++ b/kafka/consumer/kafka.py
@@ -17,7 +17,6 @@ from kafka.common import (
     OffsetOutOfRangeError, RequestTimedOutError, KafkaMessage, ConsumerTimeout,
     FailedPayloadsError, KafkaUnavailableError, KafkaConfigurationError
 )
-from kafka.util import kafka_bytestring
 
 logger = logging.getLogger(__name__)
 
@@ -193,14 +192,14 @@ class KafkaConsumer(object):
 
             # Topic name str -- all partitions
             if isinstance(arg, (six.string_types, six.binary_type)):
-                topic = kafka_bytestring(arg)
+                topic = arg
 
                 for partition in self._client.get_partition_ids_for_topic(topic):
                     self._consume_topic_partition(topic, partition)
 
             # (topic, partition [, offset]) tuple
             elif isinstance(arg, tuple):
-                topic = kafka_bytestring(arg[0])
+                topic = arg[0]
                 partition = arg[1]
                 self._consume_topic_partition(topic, partition)
                 if len(arg) == 3:
@@ -213,7 +212,7 @@ class KafkaConsumer(object):
 
                     # key can be string (a topic)
                     if isinstance(key, (six.string_types, six.binary_type)):
-                        topic = kafka_bytestring(key)
+                        topic = key
 
                         # topic: partition
                         if isinstance(value, int):
@@ -231,7 +230,7 @@ class KafkaConsumer(object):
 
                     # (topic, partition): offset
                     elif isinstance(key, tuple):
-                        topic = kafka_bytestring(key[0])
+                        topic = key[0]
                         partition = key[1]
                         self._consume_topic_partition(topic, partition)
                         self._offsets.fetch[(topic, partition)] = value
@@ -354,7 +353,7 @@ class KafkaConsumer(object):
                 self._refresh_metadata_on_error()
                 continue
 
-            topic = kafka_bytestring(resp.topic)
+            topic = resp.topic
             partition = resp.partition
             try:
                 check_error(resp)
@@ -553,7 +552,7 @@ class KafkaConsumer(object):
         if commits:
             logger.info('committing consumer offsets to group %s', self._config['group_id'])
             resps = self._client.send_offset_commit_request(
-                kafka_bytestring(self._config['group_id']), commits,
+                self._config['group_id'], commits,
                 fail_on_error=False
             )
 
@@ -577,7 +576,6 @@ class KafkaConsumer(object):
     #
 
     def _consume_topic_partition(self, topic, partition):
-        topic = kafka_bytestring(topic)
         if not isinstance(partition, int):
             raise KafkaConfigurationError('Unknown partition type (%s) '
                                           '-- expected int' % type(partition))
@@ -617,7 +615,7 @@ class KafkaConsumer(object):
         logger.info("Consumer fetching stored offsets")
         for topic_partition in self._topics:
             (resp,) = self._client.send_offset_fetch_request(
-                kafka_bytestring(self._config['group_id']),
+                self._config['group_id'],
                 [OffsetFetchRequestPayload(topic_partition[0], topic_partition[1])],
                 fail_on_error=False)
             try:
diff --git a/kafka/producer/base.py b/kafka/producer/base.py
index 3f2bba6..595ac37 100644
--- a/kafka/producer/base.py
+++ b/kafka/producer/base.py
@@ -22,7 +22,6 @@ from kafka.common import (
 )
 
 from kafka.protocol import CODEC_NONE, ALL_CODECS, create_message_set
-from kafka.util import kafka_bytestring
 
 log = logging.getLogger('kafka.producer')
 
@@ -361,7 +360,6 @@ 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):
@@ -381,10 +379,6 @@ class Producer(object):
             elif not isinstance(m, six.binary_type):
                 raise TypeError("all produce message payloads must be null or 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")
diff --git a/kafka/producer/keyed.py b/kafka/producer/keyed.py
index a5a26c9..f35aef0 100644
--- a/kafka/producer/keyed.py
+++ b/kafka/producer/keyed.py
@@ -5,7 +5,6 @@ import warnings
 
 from .base import Producer
 from ..partitioner import HashedPartitioner
-from ..util import kafka_bytestring
 
 
 log = logging.getLogger(__name__)
@@ -38,7 +37,6 @@ class KeyedProducer(Producer):
         return partitioner.partition(key)
 
     def send_messages(self, topic, key, *msg):
-        topic = kafka_bytestring(topic)
         partition = self._next_partition(topic, key)
         return self._send_messages(topic, partition, *msg, key=key)
 
diff --git a/kafka/producer/simple.py b/kafka/producer/simple.py
index 13e60d9..d8647b6 100644
--- a/kafka/producer/simple.py
+++ b/kafka/producer/simple.py
@@ -46,9 +46,6 @@ class SimpleProducer(Producer):
         return next(self.partition_cycles[topic])
 
     def send_messages(self, topic, *msg):
-        if not isinstance(topic, six.binary_type):
-            topic = topic.encode('utf-8')
-
         partition = self._next_partition(topic)
         return super(SimpleProducer, self).send_messages(
             topic, partition, *msg
diff --git a/kafka/util.py b/kafka/util.py
index e95d51d..c6e77fa 100644
--- a/kafka/util.py
+++ b/kafka/util.py
@@ -95,18 +95,6 @@ def group_by_topic_and_partition(tuples):
     return out
 
 
-def kafka_bytestring(s):
-    """
-    Takes a string or bytes instance
-    Returns bytes, encoding strings in utf-8 as necessary
-    """
-    if isinstance(s, six.binary_type):
-        return s
-    if isinstance(s, six.string_types):
-        return s.encode('utf-8')
-    raise TypeError(s)
-
-
 class ReentrantTimer(object):
     """
     A timer that can be restarted, unlike threading.Timer
diff --git a/test/test_client.py b/test/test_client.py
index a3e04f4..dd8948f 100644
--- a/test/test_client.py
+++ b/test/test_client.py
@@ -133,12 +133,12 @@ class TestKafkaClient(unittest.TestCase):
         # client loads metadata at init
         client = KafkaClient(hosts=['broker_1:4567'])
         self.assertDictEqual({
-            TopicAndPartition(b'topic_1', 0): brokers[1],
-            TopicAndPartition(b'topic_noleader', 0): None,
-            TopicAndPartition(b'topic_noleader', 1): None,
-            TopicAndPartition(b'topic_3', 0): brokers[0],
-            TopicAndPartition(b'topic_3', 1): brokers[1],
-            TopicAndPartition(b'topic_3', 2): brokers[0]},
+            TopicAndPartition('topic_1', 0): brokers[1],
+            TopicAndPartition('topic_noleader', 0): None,
+            TopicAndPartition('topic_noleader', 1): None,
+            TopicAndPartition('topic_3', 0): brokers[0],
+            TopicAndPartition('topic_3', 1): brokers[1],
+            TopicAndPartition('topic_3', 2): brokers[0]},
             client.topics_to_brokers)
 
         # if we ask for metadata explicitly, it should raise errors
@@ -150,7 +150,6 @@ class TestKafkaClient(unittest.TestCase):
 
         # This should not raise
         client.load_metadata_for_topics('topic_no_leader')
-        client.load_metadata_for_topics(b'topic_no_leader')
 
     @patch('kafka.client.BrokerConnection')
     @patch('kafka.client.KafkaProtocol')
@@ -274,10 +273,10 @@ class TestKafkaClient(unittest.TestCase):
         self.assertDictEqual({}, client.topics_to_brokers)
 
         with self.assertRaises(LeaderNotAvailableError):
-            client._get_leader_for_partition(b'topic_no_partitions', 0)
+            client._get_leader_for_partition('topic_no_partitions', 0)
 
         with self.assertRaises(UnknownTopicOrPartitionError):
-            client._get_leader_for_partition(b'topic_unknown', 0)
+            client._get_leader_for_partition('topic_unknown', 0)
 
     @patch('kafka.client.BrokerConnection')
     @patch('kafka.client.KafkaProtocol')
diff --git a/test/test_client_integration.py b/test/test_client_integration.py
index edd62da..1599006 100644
--- a/test/test_client_integration.py
+++ b/test/test_client_integration.py
@@ -29,11 +29,11 @@ class TestKafkaClientIntegration(KafkaIntegrationTestCase):
 
     @kafka_versions("all")
     def test_consume_none(self):
-        fetch = FetchRequestPayload(self.bytes_topic, 0, 0, 1024)
+        fetch = FetchRequestPayload(self.topic, 0, 0, 1024)
 
         fetch_resp, = self.client.send_fetch_request([fetch])
         self.assertEqual(fetch_resp.error, 0)
-        self.assertEqual(fetch_resp.topic, self.bytes_topic)
+        self.assertEqual(fetch_resp.topic, self.topic)
         self.assertEqual(fetch_resp.partition, 0)
 
         messages = list(fetch_resp.messages)
@@ -48,26 +48,26 @@ class TestKafkaClientIntegration(KafkaIntegrationTestCase):
 
         # ensure_topic_exists should fail with KafkaTimeoutError
         with self.assertRaises(KafkaTimeoutError):
-            self.client.ensure_topic_exists(b"this_topic_doesnt_exist", timeout=0)
+            self.client.ensure_topic_exists('this_topic_doesnt_exist', timeout=0)
 
     @kafka_versions('all')
     def test_send_produce_request_maintains_request_response_order(self):
 
-        self.client.ensure_topic_exists(b'foo')
-        self.client.ensure_topic_exists(b'bar')
+        self.client.ensure_topic_exists('foo')
+        self.client.ensure_topic_exists('bar')
 
         requests = [
             ProduceRequestPayload(
-                b'foo', 0,
+                'foo', 0,
                 [create_message(b'a'), create_message(b'b')]),
             ProduceRequestPayload(
-                b'bar', 1,
+                'bar', 1,
                 [create_message(b'a'), create_message(b'b')]),
             ProduceRequestPayload(
-                b'foo', 1,
+                'foo', 1,
                 [create_message(b'a'), create_message(b'b')]),
             ProduceRequestPayload(
-                b'bar', 0,
+                'bar', 0,
                 [create_message(b'a'), create_message(b'b')]),
         ]
 
@@ -85,12 +85,12 @@ class TestKafkaClientIntegration(KafkaIntegrationTestCase):
 
     @kafka_versions("0.8.1", "0.8.1.1", "0.8.2.1")
     def test_commit_fetch_offsets(self):
-        req = OffsetCommitRequestPayload(self.bytes_topic, 0, 42, b"metadata")
-        (resp,) = self.client.send_offset_commit_request(b"group", [req])
+        req = OffsetCommitRequestPayload(self.topic, 0, 42, 'metadata')
+        (resp,) = self.client.send_offset_commit_request('group', [req])
         self.assertEqual(resp.error, 0)
 
-        req = OffsetFetchRequestPayload(self.bytes_topic, 0)
-        (resp,) = self.client.send_offset_fetch_request(b"group", [req])
+        req = OffsetFetchRequestPayload(self.topic, 0)
+        (resp,) = self.client.send_offset_fetch_request('group', [req])
         self.assertEqual(resp.error, 0)
         self.assertEqual(resp.offset, 42)
-        self.assertEqual(resp.metadata, b"")  # Metadata isn't stored for now
+        self.assertEqual(resp.metadata, '')  # Metadata isn't stored for now
diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py
index 4cebed8..43370e2 100644
--- a/test/test_consumer_integration.py
+++ b/test/test_consumer_integration.py
@@ -41,7 +41,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase):
 
     def send_messages(self, partition, messages):
         messages = [ create_message(self.msg(str(msg))) for msg in messages ]
-        produce = ProduceRequestPayload(self.bytes_topic, partition, messages = messages)
+        produce = ProduceRequestPayload(self.topic, partition, messages = messages)
         resp, = self.client.send_produce_request([produce])
         self.assertEqual(resp.error, 0)
 
diff --git a/test/test_failover_integration.py b/test/test_failover_integration.py
index 91779d7..8c5efe2 100644
--- a/test/test_failover_integration.py
+++ b/test/test_failover_integration.py
@@ -5,7 +5,6 @@ import time
 from kafka import KafkaClient, SimpleConsumer, KeyedProducer
 from kafka.common import TopicAndPartition, FailedPayloadsError, ConnectionError
 from kafka.producer.base import Producer
-from kafka.util import kafka_bytestring
 
 from test.fixtures import ZookeeperFixture, KafkaFixture
 from test.testutil import (
@@ -165,7 +164,7 @@ class TestFailover(KafkaIntegrationTestCase):
                 key = random_string(3).encode('utf-8')
                 msg = random_string(10).encode('utf-8')
                 producer.send_messages(topic, key, msg)
-                if producer.partitioners[kafka_bytestring(topic)].partition(key) == 0:
+                if producer.partitioners[topic].partition(key) == 0:
                     recovered = True
             except (FailedPayloadsError, ConnectionError):
                 log.debug("caught exception sending message -- will retry")
@@ -203,7 +202,7 @@ class TestFailover(KafkaIntegrationTestCase):
                     break
 
     def _kill_leader(self, topic, partition):
-        leader = self.client.topics_to_brokers[TopicAndPartition(kafka_bytestring(topic), partition)]
+        leader = self.client.topics_to_brokers[TopicAndPartition(topic, partition)]
         broker = self.brokers[leader.nodeId]
         broker.close()
         return broker
diff --git a/test/test_producer_integration.py b/test/test_producer_integration.py
index ee0b2fd..d09c1af 100644
--- a/test/test_producer_integration.py
+++ b/test/test_producer_integration.py
@@ -488,7 +488,7 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
 
     def assert_produce_request(self, messages, initial_offset, message_ct,
                                partition=0):
-        produce = ProduceRequestPayload(self.bytes_topic, partition, messages=messages)
+        produce = ProduceRequestPayload(self.topic, partition, messages=messages)
 
         # There should only be one response message from the server.
         # This will throw an exception if there's more than one.
@@ -506,7 +506,7 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
         # There should only be one response message from the server.
         # This will throw an exception if there's more than one.
 
-        resp, = self.client.send_fetch_request([FetchRequestPayload(self.bytes_topic, partition, start_offset, 1024)])
+        resp, = self.client.send_fetch_request([FetchRequestPayload(self.topic, partition, start_offset, 1024)])
 
         self.assertEqual(resp.error, 0)
         self.assertEqual(resp.partition, partition)
diff --git a/test/testutil.py b/test/testutil.py
index b5b2529..5c6ea1b 100644
--- a/test/testutil.py
+++ b/test/testutil.py
@@ -12,7 +12,6 @@ from . import unittest
 
 from kafka import KafkaClient
 from kafka.common import OffsetRequestPayload
-from kafka.util import kafka_bytestring
 
 __all__ = [
     'random_string',
@@ -50,7 +49,6 @@ def get_open_port():
 class KafkaIntegrationTestCase(unittest.TestCase):
     create_client = True
     topic = None
-    bytes_topic = None
     zk = None
     server = None
 
@@ -62,7 +60,6 @@ class KafkaIntegrationTestCase(unittest.TestCase):
         if not self.topic:
             topic = "%s-%s" % (self.id()[self.id().rindex(".") + 1:], random_string(10))
             self.topic = topic
-            self.bytes_topic = topic.encode('utf-8')
 
         if self.create_client:
             self.client = KafkaClient('%s:%d' % (self.server.host, self.server.port))
@@ -81,7 +78,7 @@ class KafkaIntegrationTestCase(unittest.TestCase):
 
     def current_offset(self, topic, partition):
         try:
-            offsets, = self.client.send_offset_request([OffsetRequestPayload(kafka_bytestring(topic), partition, -1, 1)])
+            offsets, = self.client.send_offset_request([OffsetRequestPayload(topic, partition, -1, 1)])
         except:
             # XXX: We've seen some UnknownErrors here and cant debug w/o server logs
             self.zk.child.dump_logs()