Rename Assignor.metadata() -> Assignor.subscription()

Author: dpkp

kafka/coordinator/assignors/abstract.py

@@ -32,8 +32,8 @@ def assign(self, cluster, members):
         pass
 
     @abc.abstractmethod
-    def metadata(self, topics):
-        """Generate ProtocolMetadata to be submitted via JoinGroupRequest.
+    def subscription(self, topics):
+        """Generate subscription data to be submitted via JoinGroupRequest.
 
         Arguments:
             topics (set): a member's subscribed topics

kafka/coordinator/assignors/range.py

@@ -73,7 +73,7 @@ def assign(cls, cluster, group_subscriptions):
         return protocol_assignment
 
     @classmethod
-    def metadata(cls, topics):
+    def subscription(cls, topics):
         return ConsumerProtocolSubscription(cls.version, list(topics), b'')
 
     @classmethod

kafka/coordinator/assignors/roundrobin.py

@@ -92,7 +92,7 @@ def assign(cls, cluster, group_subscriptions):
         return protocol_assignment
 
     @classmethod
-    def metadata(cls, topics):
+    def subscription(cls, topics):
         return ConsumerProtocolSubscription(cls.version, list(topics), b'')
 
     @classmethod

kafka/coordinator/assignors/sticky/sticky_assignor.py

@@ -636,7 +636,7 @@ def parse_member_metadata(cls, metadata):
         )
 
     @classmethod
-    def metadata(cls, topics):
+    def subscription(cls, topics):
         return cls._metadata(topics, cls.member_assignment, cls.generation)
 
     @classmethod
@@ -645,7 +645,7 @@ def _metadata(cls, topics, member_assignment_partitions, generation=-1):
             log.debug("No member assignment available")
             user_data = b''
         else:
-            log.debug("Member assignment is available, generating the metadata: generation {}".format(cls.generation))
+            log.debug("Member assignment is available, generating subscription: generation {}".format(cls.generation))
             partitions_by_topic = defaultdict(list)
             for topic_partition in member_assignment_partitions:
                 partitions_by_topic[topic_partition.topic].append(topic_partition.partition)

kafka/coordinator/consumer.py

@@ -168,7 +168,7 @@ def group_protocols(self):
         self._joined_subscription = set(self._subscription.subscription)
         metadata_list = []
         for assignor in self.config['assignors']:
-            metadata = assignor.metadata(self._joined_subscription)
+            metadata = assignor.subscription(self._joined_subscription)
             group_protocol = (assignor.name, metadata)
             metadata_list.append(group_protocol)
         return metadata_list

test/test_assignors.py

@@ -34,8 +34,8 @@ def test_assignor_roundrobin(mocker):
     assignor = RoundRobinPartitionAssignor
 
     group_subscriptions = {
-        'C0': Subscription(assignor.metadata({'t0', 't1'}), None),
-        'C1': Subscription(assignor.metadata({'t0', 't1'}), None),
+        'C0': Subscription(assignor.subscription({'t0', 't1'}), None),
+        'C1': Subscription(assignor.subscription({'t0', 't1'}), None),
     }
 
     cluster = create_cluster(mocker, {'t0', 't1'}, topics_partitions={0, 1, 2})
@@ -56,8 +56,8 @@ def test_assignor_range(mocker):
     assignor = RangePartitionAssignor
 
     group_subscriptions = {
-        'C0': Subscription(assignor.metadata({'t0', 't1'}), None),
-        'C1': Subscription(assignor.metadata({'t0', 't1'}), None),
+        'C0': Subscription(assignor.subscription({'t0', 't1'}), None),
+        'C1': Subscription(assignor.subscription({'t0', 't1'}), None),
     }
 
     cluster = create_cluster(mocker, {'t0', 't1'}, topics_partitions={0, 1, 2})