openstack/deb-python-kafka
Code Issues Proposed changes

Sort partitions before calling partitioner (#905)

Browse Source
This commit is contained in:
ms7s
2016-12-19 20:26:03 +01:00
committed by Dana Powers
parent e8283958e4
commit 46f9b1f681
2 changed files with 8 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
kafka/partitioner/default.py
View File

@@ -14,6 +14,13 @@ class DefaultPartitioner(object):
""" """
@classmethod @classmethod
def __call__(cls, key, all_partitions, available): def __call__(cls, key, all_partitions, available):
"""
Get the partition corresponding to key
:param key: partitioning key
:param all_partitions: list of all partitions sorted by partition ID
:param available: list of available partitions in no particular order
:return: one of the values from all_partitions or available
"""
if key is None: if key is None:
if available: if available:
return random.choice(available) return random.choice(available)

2
kafka/producer/kafka.py
View File

@@ -625,7 +625,7 @@ class KafkaProducer(object):
assert partition in self._metadata.partitions_for_topic(topic), 'Unrecognized partition' assert partition in self._metadata.partitions_for_topic(topic), 'Unrecognized partition'
return partition return partition
all_partitions = list(self._metadata.partitions_for_topic(topic)) all_partitions = sorted(self._metadata.partitions_for_topic(topic))
available = list(self._metadata.available_partitions_for_topic(topic)) available = list(self._metadata.available_partitions_for_topic(topic))
return self.config['partitioner'](serialized_key, return self.config['partitioner'](serialized_key,
all_partitions, all_partitions,