(self, contexts_limit)
Fetch Consumer Group offsets from Kafka.

        These offsets are stored in the __consumer_offsets topic rather than in Zookeeper.
Fetch Consumer Group offsets from Kafka.
Fetch Consumer Group offsets from Kafka .
def _get_kafka_consumer_offsets(self, contexts_limit):
        """
        Fetch Consumer Group offsets from Kafka.

        These offsets are stored in the __consumer_offsets topic rather than in Zookeeper.
        """
        for consumer_group, topic_partitions in self._consumer_groups.items():
            if not topic_partitions:
                raise ConfigurationError(
                    'Invalid configuration - if you are collecting consumer offsets from Kafka, and your brokers are '
                    'older than 0.10.2, then you _must_ specify consumer groups and their topics. Older brokers lack '
                    'the necessary protocol support to determine which topics a consumer is consuming. See KIP-88 for '
                    'details.'
                )
            try:  # catch exceptions on a group-by-group basis so that if one fails we still fetch the other groups
                for topic, partitions in topic_partitions.items():
                    if not partitions:
                        # If partitions omitted, then we assume the group is consuming all partitions for the topic.
                        # Fetch consumer offsets even for unavailable partitions because those will be valid once the
                        # partition finishes leader failover.
                        topic_partitions[topic] = self.kafka_client.cluster.partitions_for_topic(topic)

                coordinator_id = self._get_group_coordinator(consumer_group)
                if coordinator_id is not None:
                    # Kafka protocol uses OffsetFetchRequests to retrieve consumer offsets:
                    # https://kafka.apache.org/protocol#The_Messages_OffsetFetch
                    # https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetFetchRequest
                    request = OffsetFetchRequest[1](consumer_group, list(topic_partitions.items()))
                    response = self._make_blocking_req(request, node_id=coordinator_id)
                    for (topic, partition_offsets) in response.topics:
                        for partition, offset, _metadata, error_code in partition_offsets:
                            error_type = kafka_errors.for_code(error_code)
                            # 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 or error_type is not kafka_errors.NoError:
                                self.kafka_client.cluster.request_update()  # force metadata update on next poll()
                                continue
                            key = (consumer_group, topic, partition)
                            self._kafka_consumer_offsets[key] = offset

                            if len(self._kafka_consumer_offsets) >= contexts_limit:
                                self.warning("Context limit reached. Skipping kafka consumer offsets collection.")
                                return
                else:
                    self.log.info("unable to find group coordinator for %s", consumer_group)
            except Exception:
                self.log.exception('Could not read consumer offsets from Kafka for group: %s', consumer_group)
def _get_kafka_consumer_offsets ( self , contexts_limit ) : for consumer_group , topic_partitions in self . _consumer_groups . items ( ) : if not topic_partitions : raise ConfigurationError ( 'Invalid configuration - if you are collecting consumer offsets from Kafka, and your brokers are ' 'older than 0.10.2, then you _must_ specify consumer groups and their topics. Older brokers lack ' 'the necessary protocol support to determine which topics a consumer is consuming. See KIP-88 for ' 'details.' ) try : # catch exceptions on a group-by-group basis so that if one fails we still fetch the other groups for topic , partitions in topic_partitions . items ( ) : if not partitions : # If partitions omitted, then we assume the group is consuming all partitions for the topic. # Fetch consumer offsets even for unavailable partitions because those will be valid once the # partition finishes leader failover. topic_partitions [ topic ] = self . kafka_client . cluster . partitions_for_topic ( topic ) coordinator_id = self . _get_group_coordinator ( consumer_group ) if coordinator_id is not None : # Kafka protocol uses OffsetFetchRequests to retrieve consumer offsets: # https://kafka.apache.org/protocol#The_Messages_OffsetFetch # https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetFetchRequest request = OffsetFetchRequest [ 1 ] ( consumer_group , list ( topic_partitions . items ( ) ) ) response = self . _make_blocking_req ( request , node_id = coordinator_id ) for ( topic , partition_offsets ) in response . topics : for partition , offset , _metadata , error_code in partition_offsets : error_type = kafka_errors . for_code ( error_code ) # 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 or error_type is not kafka_errors . NoError : self . kafka_client . cluster . request_update ( ) # force metadata update on next poll() continue key = ( consumer_group , topic , partition ) self . _kafka_consumer_offsets [ key ] = offset if len ( self . _kafka_consumer_offsets ) >= contexts_limit : self . warning ( "Context limit reached. Skipping kafka consumer offsets collection." ) return else : self . log . info ( "unable to find group coordinator for %s" , consumer_group ) except Exception : self . log . exception ( 'Could not read consumer offsets from Kafka for group: %s' , consumer_group )