KIP-345 Consumer group static membership

kafka/admin/client.py

@@ -13,7 +13,7 @@
 from kafka.admin.acl_resource import ACLOperation, ACLPermissionType, ACLFilter, ACL, ResourcePattern, ResourceType, \
     ACLResourcePatternType
 from kafka.client_async import KafkaClient, selectors
-from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment, ConsumerProtocol
+from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata_v0, ConsumerProtocolMemberAssignment_v0, ConsumerProtocol_v0
 import kafka.errors as Errors
 from kafka.errors import (
     IncompatibleBrokerVersion, KafkaConfigurationError, UnknownTopicOrPartitionError,
@@ -1242,7 +1242,7 @@ def _describe_consumer_groups_process_response(self, response):
                     for (described_group_information, group_information_name, group_information_field) in zip(described_group, described_groups_field_schema.names, described_groups_field_schema.fields):
                         if group_information_name == 'protocol_type':
                             protocol_type = described_group_information
-                            protocol_type_is_consumer = (protocol_type == ConsumerProtocol.PROTOCOL_TYPE or not protocol_type)
+                            protocol_type_is_consumer = (protocol_type == ConsumerProtocol_v0.PROTOCOL_TYPE or not protocol_type)
                         if isinstance(group_information_field, Array):
                             member_information_list = []
                             member_schema = group_information_field.array_of
@@ -1251,9 +1251,9 @@ def _describe_consumer_groups_process_response(self, response):
                                 for (member, member_field, member_name)  in zip(members, member_schema.fields, member_schema.names):
                                     if protocol_type_is_consumer:
                                         if member_name == 'member_metadata' and member:
-                                            member_information.append(ConsumerProtocolMemberMetadata.decode(member))
+                                            member_information.append(ConsumerProtocolMemberMetadata_v0.decode(member))
                                         elif member_name == 'member_assignment' and member:
-                                            member_information.append(ConsumerProtocolMemberAssignment.decode(member))
+                                            member_information.append(ConsumerProtocolMemberAssignment_v0.decode(member))
                                         else:
                                             member_information.append(member)
                                 member_info_tuple = MemberInformation._make(member_information)

kafka/consumer/group.py

@@ -2,6 +2,7 @@
 
 import copy
 import logging
+import re
 import socket
 import time
 
@@ -57,6 +58,14 @@ class KafkaConsumer(six.Iterator):
             committing offsets. If None, auto-partition assignment (via
             group coordinator) and offset commits are disabled.
             Default: None
+        group_instance_id (str): A unique identifier of the consumer instance
+            provided by end user. Only non-empty strings are permitted. If set,
+            the consumer is treated as a static member, which means that only
+            one instance with this ID is allowed in the consumer group at any
+            time. This can be used in combination with a larger session timeout
+            to avoid group rebalances caused by transient unavailability (e.g.
+            process restarts). If not set, the consumer will join the group as
+            a dynamic member, which is the traditional behavior. Default: None
         key_deserializer (callable): Any callable that takes a
             raw message key and returns a deserialized key.
         value_deserializer (callable): Any callable that takes a
@@ -276,6 +285,7 @@ class KafkaConsumer(six.Iterator):
         'bootstrap_servers': 'localhost',
         'client_id': 'kafka-python-' + __version__,
         'group_id': None,
+        'group_instance_id': None,
         'key_deserializer': None,
         'value_deserializer': None,
         'enable_incremental_fetch_sessions': True,
@@ -408,6 +418,10 @@ def __init__(self, *topics, **configs):
                     "Request timeout (%s) must be larger than session timeout (%s)" %
                     (self.config['request_timeout_ms'], self.config['session_timeout_ms']))
 
+        if self.config['group_instance_id'] is not None:
+            if self.config['group_id'] is None:
+                raise KafkaConfigurationError("group_instance_id requires group_id")
+
         self._subscription = SubscriptionState(self.config['auto_offset_reset'])
         self._fetcher = Fetcher(
             self._client, self._subscription, metrics=self._metrics, **self.config)
@@ -423,6 +437,16 @@ def __init__(self, *topics, **configs):
             self._subscription.subscribe(topics=topics)
             self._client.set_topics(topics)
 
+    def _validate_group_instance_id(self, group_instance_id):
+        if not group_instance_id or not isinstance(group_instance_id, str):
+            raise KafkaConfigurationError("group_instance_id must be non-empty string")
+        if group_instance_id in (".", ".."):
+            raise KafkaConfigurationError("group_instance_id cannot be \".\" or \"..\"")
+        if len(group_instance_id) > 249:
+            raise KafkaConfigurationError("group_instance_id can't be longer than 249 characters")
+        if not re.match(r'^[A-Za-z0-9\.\_\-]+$', group_instance_id):
+            raise KafkaConfigurationError("group_instance_id is illegal: it contains a character other than ASCII alphanumerics, '.', '_' and '-'")
+
     def bootstrap_connected(self):
         """Return True if the bootstrap is connected."""
         return self._client.bootstrap_connected()

kafka/coordinator/assignors/abstract.py

@@ -23,8 +23,9 @@ def assign(self, cluster, members):
 
         Arguments:
             cluster (ClusterMetadata): metadata for use in assignment
-            members (dict of {member_id: MemberMetadata}): decoded metadata for
-                each member in the group.
+            members (dict of {member_id: Subscription}): decoded metadata
+                for each member in the group, including group_instance_id
+                when available.
 
         Returns:
             dict: {member_id: MemberAssignment}

kafka/coordinator/assignors/range.py

@@ -1,12 +1,13 @@
 from __future__ import absolute_import
 
 import collections
+import itertools
 import logging
 
 from kafka.vendor import six
 
 from kafka.coordinator.assignors.abstract import AbstractPartitionAssignor
-from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment
+from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata_v0, ConsumerProtocolMemberAssignment_v0
 
 log = logging.getLogger(__name__)
 
@@ -32,45 +33,49 @@ class RangePartitionAssignor(AbstractPartitionAssignor):
     version = 0
 
     @classmethod
-    def assign(cls, cluster, member_metadata):
+    def assign(cls, cluster, group_subscriptions):
         consumers_per_topic = collections.defaultdict(list)
-        for member, metadata in six.iteritems(member_metadata):
-            for topic in metadata.subscription:
-                consumers_per_topic[topic].append(member)
+        for member_id, subscription in six.iteritems(group_subscriptions):
+            for topic in subscription.topics:
+                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 member_metadata:
-            protocol_assignment[member_id] = ConsumerProtocolMemberAssignment(
+        for member_id in group_subscriptions:
+            protocol_assignment[member_id] = ConsumerProtocolMemberAssignment_v0(
                 cls.version,
                 sorted(assignment[member_id].items()),
                 b'')
         return protocol_assignment
 
     @classmethod
     def metadata(cls, topics):
-        return ConsumerProtocolMemberMetadata(cls.version, list(topics), b'')
+        return ConsumerProtocolMemberMetadata_v0(cls.version, list(topics), b'')
 
     @classmethod
     def on_assignment(cls, assignment):

kafka/coordinator/assignors/roundrobin.py

@@ -7,7 +7,7 @@
 from kafka.vendor import six
 
 from kafka.coordinator.assignors.abstract import AbstractPartitionAssignor
-from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment
+from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata_v0, ConsumerProtocolMemberAssignment_v0
 from kafka.structs import TopicPartition
 
 log = logging.getLogger(__name__)
@@ -49,10 +49,10 @@ class RoundRobinPartitionAssignor(AbstractPartitionAssignor):
     version = 0
 
     @classmethod
-    def assign(cls, cluster, member_metadata):
+    def assign(cls, cluster, group_subscriptions):
         all_topics = set()
-        for metadata in six.itervalues(member_metadata):
-            all_topics.update(metadata.subscription)
+        for subscription in six.itervalues(group_subscriptions):
+            all_topics.update(subscription.topics)
 
         all_topic_partitions = []
         for topic in all_topics:
@@ -67,29 +67,34 @@ def assign(cls, cluster, member_metadata):
         # construct {member_id: {topic: [partition, ...]}}
         assignment = collections.defaultdict(lambda: collections.defaultdict(list))
 
-        member_iter = itertools.cycle(sorted(member_metadata.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
             # each topic in all_topic_partitions is in at least one member
             # subscription; otherwise this could yield an infinite loop
-            while partition.topic not in member_metadata[member_id].subscription:
+            while partition.topic not in group_subscriptions[member_id].topics:
                 member_id = next(member_iter)
             assignment[member_id][partition.topic].append(partition.partition)
 
         protocol_assignment = {}
-        for member_id in member_metadata:
-            protocol_assignment[member_id] = ConsumerProtocolMemberAssignment(
+        for member_id in group_subscriptions:
+            protocol_assignment[member_id] = ConsumerProtocolMemberAssignment_v0(
                 cls.version,
                 sorted(assignment[member_id].items()),
                 b'')
         return protocol_assignment
 
     @classmethod
     def metadata(cls, topics):
-        return ConsumerProtocolMemberMetadata(cls.version, list(topics), b'')
+        return ConsumerProtocolMemberMetadata_v0(cls.version, list(topics), b'')
 
     @classmethod
     def on_assignment(cls, assignment):

kafka/coordinator/assignors/sticky/sticky_assignor.py

@@ -5,7 +5,7 @@
 from kafka.coordinator.assignors.abstract import AbstractPartitionAssignor
 from kafka.coordinator.assignors.sticky.partition_movements import PartitionMovements
 from kafka.coordinator.assignors.sticky.sorted_set import SortedSet
-from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment
+from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata_v0, ConsumerProtocolMemberAssignment_v0
 from kafka.coordinator.protocol import Schema
 from kafka.protocol.struct import Struct
 from kafka.protocol.types import String, Array, Int32
@@ -66,6 +66,7 @@ class StickyAssignorUserDataV1(Struct):
 
 class StickyAssignmentExecutor:
     def __init__(self, cluster, members):
+        # a mapping of member_id => StickyAssignorMemberMetadataV1
         self.members = members
         # a mapping between consumers and their assigned partitions that is updated during assignment procedure
         self.current_assignment = defaultdict(list)
@@ -603,7 +604,7 @@ def assign(cls, cluster, members):
 
         assignment = {}
         for member_id in members:
-            assignment[member_id] = ConsumerProtocolMemberAssignment(
+            assignment[member_id] = ConsumerProtocolMemberAssignment_v0(
                 cls.version, sorted(executor.get_final_assignment(member_id)), b''
             )
         return assignment
@@ -625,24 +626,24 @@ def parse_member_metadata(cls, metadata):
         user_data = metadata.user_data
         if not user_data:
             return StickyAssignorMemberMetadataV1(
-                partitions=[], generation=cls.DEFAULT_GENERATION_ID, subscription=metadata.subscription
+                partitions=[], generation=cls.DEFAULT_GENERATION_ID, subscription=metadata.topics
             )
 
         try:
             decoded_user_data = StickyAssignorUserDataV1.decode(user_data)
-        except Exception as e:
+        except Exception:
             # ignore the consumer's previous assignment if it cannot be parsed
-            log.error("Could not parse member data", e)     # pylint: disable=logging-too-many-args
+            log.exception("Could not parse member data")
             return StickyAssignorMemberMetadataV1(
-                partitions=[], generation=cls.DEFAULT_GENERATION_ID, subscription=metadata.subscription
+                partitions=[], generation=cls.DEFAULT_GENERATION_ID, subscription=metadata.topics
             )
 
         member_partitions = []
         for topic, partitions in decoded_user_data.previous_assignment:     # pylint: disable=no-member
             member_partitions.extend([TopicPartition(topic, partition) for partition in partitions])
         return StickyAssignorMemberMetadataV1(
             # pylint: disable=no-member
-            partitions=member_partitions, generation=decoded_user_data.generation, subscription=metadata.subscription
+            partitions=member_partitions, generation=decoded_user_data.generation, subscription=metadata.topics
         )
 
     @classmethod
@@ -661,7 +662,7 @@ def _metadata(cls, topics, member_assignment_partitions, generation=-1):
                 partitions_by_topic[topic_partition.topic].append(topic_partition.partition)
             data = StickyAssignorUserDataV1(list(partitions_by_topic.items()), generation)
             user_data = data.encode()
-        return ConsumerProtocolMemberMetadata(cls.version, list(topics), user_data)
+        return ConsumerProtocolMemberMetadata_v0(cls.version, list(topics), user_data)
 
     @classmethod
     def on_assignment(cls, assignment):