(self, consumer_group, response)
Callback that parses an OffsetFetchResponse and saves it to the consumer_offsets dict.

        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.
Callback that parses an OffsetFetchResponse and saves it to the consumer_offsets dict.
Callback that parses an OffsetFetchResponse and saves it to the consumer_offsets dict .
def _single_group_offsets_callback(self, consumer_group, response):
        """Callback that parses an OffsetFetchResponse and saves it to the consumer_offsets dict.

        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.
        """
        single_group_offsets = self.kafka_client._list_consumer_group_offsets_process_response(response)
        for (topic, partition), (offset, _metadata) in single_group_offsets.items():
            # If the OffsetFetchRequest explicitly specified partitions, the offset could returned as -1, meaning there
            # is no recorded offset for that partition... for example, if the partition doesn't exist in the cluster.
            # So ignore it.
            if offset == -1:
                self.kafka_client._client.cluster.request_update()  # force metadata update on next poll()
                continue
            key = (consumer_group, topic, partition)
            self._consumer_offsets[key] = offset
def _single_group_offsets_callback ( self , consumer_group , response ) : single_group_offsets = self . kafka_client . _list_consumer_group_offsets_process_response ( response ) for ( topic , partition ) , ( offset , _metadata ) in single_group_offsets . items ( ) : # If the OffsetFetchRequest explicitly specified partitions, the offset could returned as -1, meaning there # is no recorded offset for that partition... for example, if the partition doesn't exist in the cluster. # So ignore it. if offset == - 1 : self . kafka_client . _client . cluster . request_update ( ) # force metadata update on next poll() continue key = ( consumer_group , topic , partition ) self . _consumer_offsets [ key ] = offset