Rename ConsumerProtocolMemberMetadata field subscription -> topics

Author: dpkp

kafka/coordinator/assignors/range.py

@@ -35,7 +35,7 @@ class RangePartitionAssignor(AbstractPartitionAssignor):
     def assign(cls, cluster, group_subscriptions):
         consumers_per_topic = collections.defaultdict(list)
         for member, subscription in six.iteritems(group_subscriptions):
-            for topic in subscription.subscription:
+            for topic in subscription.topics:
                 consumers_per_topic[topic].append(member)
 
         # construct {member_id: {topic: [partition, ...]}}

kafka/coordinator/assignors/roundrobin.py

@@ -52,7 +52,7 @@ class RoundRobinPartitionAssignor(AbstractPartitionAssignor):
     def assign(cls, cluster, group_subscriptions):
         all_topics = set()
         for subscription in six.itervalues(group_subscriptions):
-            all_topics.update(subscription.subscription)
+            all_topics.update(subscription.topics)
 
         all_topic_partitions = []
         for topic in all_topics:
@@ -75,7 +75,7 @@ def assign(cls, cluster, group_subscriptions):
             # 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 group_subscriptions[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)
 

kafka/coordinator/assignors/sticky/sticky_assignor.py

@@ -625,7 +625,7 @@ 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:
@@ -634,15 +634,15 @@ def parse_member_metadata(cls, metadata):
             # 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
             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

kafka/coordinator/consumer.py

@@ -342,7 +342,7 @@ def _perform_assignment(self, leader_id, assignment_strategy, members):
                 member.group_instance_id
             )
             member_subscriptions[member.member_id] = subscription
-            all_subscribed_topics.update(subscription.subscription)
+            all_subscribed_topics.update(subscription.topics)
 
         # the leader will begin watching for changes to any of the topics
         # the group is interested in, which ensures that all metadata changes

kafka/coordinator/protocol.py

@@ -8,7 +8,7 @@
 class ConsumerProtocolMemberMetadata_v0(Struct):
     SCHEMA = Schema(
         ('version', Int16),
-        ('subscription', Array(String('utf-8'))),
+        ('topics', Array(String('utf-8'))),
         ('user_data', Bytes))
 
 

kafka/coordinator/subscription.py

@@ -16,8 +16,11 @@ def user_data(self):
         return self._metadata.user_data
 
     @property
-    def subscription(self):
-        return self._metadata.subscription
+    def topics(self):
+        return self._metadata.topics
+
+    # Alias for old interface / name
+    subscription = topics
 
     @property
     def group_instance_id(self):

test/integration/test_admin_integration.py

@@ -231,7 +231,7 @@ def consumer_thread(i, group_id):
             else:
                 assert(len(consumer_group.members) == 1)
             for member in consumer_group.members:
-                    assert(member.member_metadata.subscription[0] == topic)
+                    assert(member.member_metadata.topics[0] == topic)
                     assert(member.member_assignment.assignment[0][0] == topic)
             consumer_groups.add(consumer_group.group)
         assert(sorted(list(consumer_groups)) == group_id_list)