(self, consumer_group, response)
Callback that takes a FindCoordinatorResponse and issues an OffsetFetchRequest for the group.

        consumer_group must be manually passed in because it is not present in the response, but we need it in order to
        associate these offsets to the proper consumer group.

        The OffsetFetchRequest is scoped to the topics and partitions that are specified in the check config. If
        topics are unspecified, it will fetch all known offsets for that consumer group. Similarly, if the partitions
        are unspecified for a topic listed in the config, offsets are fetched for all the partitions within that topic.
Callback that takes a FindCoordinatorResponse and issues an OffsetFetchRequest for the group.
Callback that takes a FindCoordinatorResponse and issues an OffsetFetchRequest for the group .
def _find_coordinator_callback(self, consumer_group, response):
        """Callback that takes a FindCoordinatorResponse and issues an OffsetFetchRequest for the group.

        consumer_group must be manually passed in because it is not present in the response, but we need it in order to
        associate these offsets to the proper consumer group.

        The OffsetFetchRequest is scoped to the topics and partitions that are specified in the check config. If
        topics are unspecified, it will fetch all known offsets for that consumer group. Similarly, if the partitions
        are unspecified for a topic listed in the config, offsets are fetched for all the partitions within that topic.
        """
        coordinator_id = self.kafka_client._find_coordinator_id_process_response(response)
        topics = self._consumer_groups[consumer_group]
        if not topics:
            topic_partitions = None  # None signals to fetch all known offsets for the consumer group
        else:
            # transform [("t1", [1, 2])] into [TopicPartition("t1", 1), TopicPartition("t1", 2)]
            topic_partitions = []
            for topic, partitions in topics.items():
                if not partitions:  # If partitions aren't specified, fetch all partitions in the topic
                    partitions = self.kafka_client._client.cluster.partitions_for_topic(topic)
                topic_partitions.extend([TopicPartition(topic, p) for p in partitions])
        single_group_offsets_future = self.kafka_client._list_consumer_group_offsets_send_request(
            group_id=consumer_group, group_coordinator_id=coordinator_id, partitions=topic_partitions
        )
        single_group_offsets_future.add_callback(self._single_group_offsets_callback, consumer_group)
        self._consumer_futures.append(single_group_offsets_future)
def _find_coordinator_callback ( self , consumer_group , response ) : coordinator_id = self . kafka_client . _find_coordinator_id_process_response ( response ) topics = self . _consumer_groups [ consumer_group ] if not topics : topic_partitions = None # None signals to fetch all known offsets for the consumer group else : # transform [("t1", [1, 2])] into [TopicPartition("t1", 1), TopicPartition("t1", 2)] topic_partitions = [ ] for topic , partitions in topics . items ( ) : if not partitions : # If partitions aren't specified, fetch all partitions in the topic partitions = self . kafka_client . _client . cluster . partitions_for_topic ( topic ) topic_partitions . extend ( [ TopicPartition ( topic , p ) for p in partitions ] ) single_group_offsets_future = self . kafka_client . _list_consumer_group_offsets_send_request ( group_id = consumer_group , group_coordinator_id = coordinator_id , partitions = topic_partitions ) single_group_offsets_future . add_callback ( self . _single_group_offsets_callback , consumer_group ) self . _consumer_futures . append ( single_group_offsets_future )