(self)
Fetch Consumer Group offsets from Kafka.

        Also fetch consumer_groups, topics, and partitions if not already specified.

        For speed, all the brokers are queried in parallel using callbacks.
        The callback flow is:
            A: When fetching all groups ('monitor_unlisted_consumer_groups' is True):
                1. Issue a ListGroupsRequest to every broker
                2. Attach a callback to each ListGroupsRequest that issues OffsetFetchRequests for every group.
                   Note: Because a broker only returns groups for which it is the coordinator, as an optimization we
                   skip the FindCoordinatorRequest
            B: When fetching only listed groups:
                1. Issue a FindCoordintorRequest for each group
                2. Attach a callback to each FindCoordinatorResponse that issues OffsetFetchRequests for that group
            Both:
                3. Attach a callback to each OffsetFetchRequest that parses the response
                   and saves the consumer group's offsets
Fetch Consumer Group offsets from Kafka.
Fetch Consumer Group offsets from Kafka .
def _get_consumer_offsets(self):
        """Fetch Consumer Group offsets from Kafka.

        Also fetch consumer_groups, topics, and partitions if not already specified.

        For speed, all the brokers are queried in parallel using callbacks.
        The callback flow is:
            A: When fetching all groups ('monitor_unlisted_consumer_groups' is True):
                1. Issue a ListGroupsRequest to every broker
                2. Attach a callback to each ListGroupsRequest that issues OffsetFetchRequests for every group.
                   Note: Because a broker only returns groups for which it is the coordinator, as an optimization we
                   skip the FindCoordinatorRequest
            B: When fetching only listed groups:
                1. Issue a FindCoordintorRequest for each group
                2. Attach a callback to each FindCoordinatorResponse that issues OffsetFetchRequests for that group
            Both:
                3. Attach a callback to each OffsetFetchRequest that parses the response
                   and saves the consumer group's offsets
        """
        # Store the list of futures on the object because some of the callbacks create/store additional futures and they
        # don't have access to variables scoped to this method, only to the object scope
        self._consumer_futures = []

        if self._monitor_unlisted_consumer_groups:
            for broker in self.kafka_client._client.cluster.brokers():
                list_groups_future = self.kafka_client._list_consumer_groups_send_request(broker.nodeId)
                list_groups_future.add_callback(self._list_groups_callback, broker.nodeId)
                self._consumer_futures.append(list_groups_future)
        elif self._consumer_groups:
            self.validate_consumer_groups()
            for consumer_group in self._consumer_groups:
                find_coordinator_future = self.kafka_client._find_coordinator_id_send_request(consumer_group)
                find_coordinator_future.add_callback(self._find_coordinator_callback, consumer_group)
                self._consumer_futures.append(find_coordinator_future)
        else:
            raise ConfigurationError(
                "Cannot fetch consumer offsets because no consumer_groups are specified and "
                "monitor_unlisted_consumer_groups is %s." % self._monitor_unlisted_consumer_groups
            )

        # Loop until all futures resolved.
        self.kafka_client._wait_for_futures(self._consumer_futures)
        del self._consumer_futures
def _get_consumer_offsets ( self ) : # Store the list of futures on the object because some of the callbacks create/store additional futures and they # don't have access to variables scoped to this method, only to the object scope self . _consumer_futures = [ ] if self . _monitor_unlisted_consumer_groups : for broker in self . kafka_client . _client . cluster . brokers ( ) : list_groups_future = self . kafka_client . _list_consumer_groups_send_request ( broker . nodeId ) list_groups_future . add_callback ( self . _list_groups_callback , broker . nodeId ) self . _consumer_futures . append ( list_groups_future ) elif self . _consumer_groups : self . validate_consumer_groups ( ) for consumer_group in self . _consumer_groups : find_coordinator_future = self . kafka_client . _find_coordinator_id_send_request ( consumer_group ) find_coordinator_future . add_callback ( self . _find_coordinator_callback , consumer_group ) self . _consumer_futures . append ( find_coordinator_future ) else : raise ConfigurationError ( "Cannot fetch consumer offsets because no consumer_groups are specified and " "monitor_unlisted_consumer_groups is %s." % self . _monitor_unlisted_consumer_groups ) # Loop until all futures resolved. self . kafka_client . _wait_for_futures ( self . _consumer_futures ) del self . _consumer_futures