(self, broker_id, response)
Callback that takes a ListGroupsResponse and issues an OffsetFetchRequest for each group.

        broker_id must be manually passed in because it is not present in the response. Keeping track of the broker that
        gave us this response lets us skip issuing FindCoordinatorRequests because Kafka brokers only include
        consumer groups in their ListGroupsResponse when they are the coordinator for that group.
Callback that takes a ListGroupsResponse and issues an OffsetFetchRequest for each group.
Callback that takes a ListGroupsResponse and issues an OffsetFetchRequest for each group .
def _list_groups_callback(self, broker_id, response):
        """Callback that takes a ListGroupsResponse and issues an OffsetFetchRequest for each group.

        broker_id must be manually passed in because it is not present in the response. Keeping track of the broker that
        gave us this response lets us skip issuing FindCoordinatorRequests because Kafka brokers only include
        consumer groups in their ListGroupsResponse when they are the coordinator for that group.
        """
        for consumer_group, group_type in self.kafka_client._list_consumer_groups_process_response(response):
            # consumer groups from Kafka < 0.9 that store their offset in Kafka don't use Kafka for group-coordination
            # so their group_type is empty
            if group_type in ('consumer', ''):
                single_group_offsets_future = self.kafka_client._list_consumer_group_offsets_send_request(
                    group_id=consumer_group, group_coordinator_id=broker_id
                )
                single_group_offsets_future.add_callback(self._single_group_offsets_callback, consumer_group)
                self._consumer_futures.append(single_group_offsets_future)
def _list_groups_callback ( self , broker_id , response ) : for consumer_group , group_type in self . kafka_client . _list_consumer_groups_process_response ( response ) : # consumer groups from Kafka < 0.9 that store their offset in Kafka don't use Kafka for group-coordination # so their group_type is empty if group_type in ( 'consumer' , '' ) : single_group_offsets_future = self . kafka_client . _list_consumer_group_offsets_send_request ( group_id = consumer_group , group_coordinator_id = broker_id ) single_group_offsets_future . add_callback ( self . _single_group_offsets_callback , consumer_group ) self . _consumer_futures . append ( single_group_offsets_future )