openstack/deb-python-kafka
Code Issues Proposed changes

Added raise of TypeError for non bytes key

Browse Source
This commit is contained in:
Lou Marvin Caraig
2014-11-26 13:05:23 +01:00
parent 664240ab4d
commit 2716d06ea0
1 changed files with 6 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
kafka/producer/base.py
View File

@@ -172,6 +172,8 @@ class Producer(object):
return self._send_messages(topic, partition, *msg) return self._send_messages(topic, partition, *msg)
def _send_messages(self, topic, partition, *msg, **kwargs): def _send_messages(self, topic, partition, *msg, **kwargs):
key = kwargs.pop('key', None)
# Guarantee that msg is actually a list or tuple (should always be true) # Guarantee that msg is actually a list or tuple (should always be true)
if not isinstance(msg, (list, tuple)): if not isinstance(msg, (list, tuple)):
raise TypeError("msg is not a list or tuple!") raise TypeError("msg is not a list or tuple!")
@@ -180,7 +182,10 @@ class Producer(object):
if any(not isinstance(m, six.binary_type) for m in msg): if any(not isinstance(m, six.binary_type) for m in msg):
raise TypeError("all produce message payloads must be type bytes") raise TypeError("all produce message payloads must be type bytes")
key = kwargs.pop('key', None) # 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")
if self.async: if self.async:
for m in msg: for m in msg:
self.queue.put((TopicAndPartition(topic, partition), m, key)) self.queue.put((TopicAndPartition(topic, partition), m, key))