Sort partitions before calling partitioner (dpkp#905)

52388fa · Dec 19, 2016 · 46f9b1f · 46f9b1f
1 parent e828395
commit 46f9b1f
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 1 deletion.
diff --git 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
@@ -625,7 +625,7 @@ def _partition(self, topic, partition, key, value,
             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,