summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorms7s <martin.sucha@exponea.com>2016-12-19 20:26:03 +0100
committerDana Powers <dana.powers@gmail.com>2016-12-19 11:26:03 -0800
commit46f9b1f681e8e999fbe6a1704c65cc35dca38f4c (patch)
tree432ce7149b3b83a92932bfe6c2e7c36d6d60c304
parente8283958e42047a31bc914fe53b2060fa5e4481b (diff)
downloadkafka-python-46f9b1f681e8e999fbe6a1704c65cc35dca38f4c.tar.gz
Sort partitions before calling partitioner (#905)
-rw-r--r--kafka/partitioner/default.py7
-rw-r--r--kafka/producer/kafka.py2
2 files changed, 8 insertions, 1 deletions
diff --git a/kafka/partitioner/default.py b/kafka/partitioner/default.py
index 79205b6..087166c 100644
--- a/kafka/partitioner/default.py
+++ b/kafka/partitioner/default.py
@@ -14,6 +14,13 @@ class DefaultPartitioner(object):
"""
@classmethod
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 available:
return random.choice(available)
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 17f27ab..1d943c6 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -625,7 +625,7 @@ class KafkaProducer(object):
assert partition in self._metadata.partitions_for_topic(topic), 'Unrecognized 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))
return self.config['partitioner'](serialized_key,
all_partitions,