Use group_instance_id in range and roundrobin assignors

Author: dpkp

kafka/coordinator/assignors/range.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 
 import collections
+import itertools
 import logging
 
 from kafka.vendor import six
@@ -34,31 +35,35 @@ class RangePartitionAssignor(AbstractPartitionAssignor):
     @classmethod
     def assign(cls, cluster, group_subscriptions):
         consumers_per_topic = collections.defaultdict(list)
-        for member, subscription in six.iteritems(group_subscriptions):
+        for member_id, subscription in six.iteritems(group_subscriptions):
             for topic in subscription.topics:
-                consumers_per_topic[topic].append(member)
+                consumers_per_topic[topic].append((subscription.group_instance_id, member_id))
 
         # construct {member_id: {topic: [partition, ...]}}
         assignment = collections.defaultdict(dict)
 
+        for topic in consumers_per_topic:
+            # group by static members (True) v dynamic members (False)
+            grouped = {k: list(g) for k, g in itertools.groupby(consumers_per_topic[topic], key=lambda ids: ids[0] is not None)}
+            consumers_per_topic[topic] = sorted(grouped.get(True, [])) + sorted(grouped.get(False, [])) # sorted static members first, then sorted dynamic
+
         for topic, consumers_for_topic in six.iteritems(consumers_per_topic):
             partitions = cluster.partitions_for_topic(topic)
             if partitions is None:
                 log.warning('No partition metadata for topic %s', topic)
                 continue
             partitions = sorted(partitions)
-            consumers_for_topic.sort()
 
             partitions_per_consumer = len(partitions) // len(consumers_for_topic)
             consumers_with_extra = len(partitions) % len(consumers_for_topic)
 
-            for i, member in enumerate(consumers_for_topic):
+            for i, (_group_instance_id, member_id) in enumerate(consumers_for_topic):
                 start = partitions_per_consumer * i
                 start += min(i, consumers_with_extra)
                 length = partitions_per_consumer
                 if not i + 1 > consumers_with_extra:
                     length += 1
-                assignment[member][topic] = partitions[start:start+length]
+                assignment[member_id][topic] = partitions[start:start+length]
 
         protocol_assignment = {}
         for member_id in group_subscriptions:

kafka/coordinator/assignors/roundrobin.py

@@ -67,9 +67,14 @@ def assign(cls, cluster, group_subscriptions):
         # construct {member_id: {topic: [partition, ...]}}
         assignment = collections.defaultdict(lambda: collections.defaultdict(list))
 
-        member_iter = itertools.cycle(sorted(group_subscriptions.keys()))
+        # Sort static and dynamic members separately to maintain stable static assignments
+        ungrouped = [(subscription.group_instance_id, member_id) for member_id, subscription in six.iteritems(group_subscriptions)]
+        grouped = {k: list(g) for k, g in itertools.groupby(ungrouped, key=lambda ids: ids[0] is not None)}
+        member_list = sorted(grouped.get(True, [])) + sorted(grouped.get(False, [])) # sorted static members first, then sorted dynamic
+        member_iter = itertools.cycle(member_list)
+
         for partition in all_topic_partitions:
-            member_id = next(member_iter)
+            _group_instance_id, member_id = next(member_iter)
 
             # Because we constructed all_topic_partitions from the set of
             # member subscribed topics, we should be safe assuming that