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

        Also fetch consumer_groups, topics, and partitions if not
        already specified in consumer_groups.
Fetch Consumer Group offsets from Zookeeper.
Fetch Consumer Group offsets from Zookeeper .
def _get_zk_consumer_offsets(self, contexts_limit):
        """
        Fetch Consumer Group offsets from Zookeeper.

        Also fetch consumer_groups, topics, and partitions if not
        already specified in consumer_groups.
        """
        # Construct the Zookeeper path pattern
        # /consumers/[groupId]/offsets/[topic]/[partitionId]
        zk_path_consumer = '/consumers/'
        zk_path_topic_tmpl = zk_path_consumer + '{group}/offsets/'
        zk_path_partition_tmpl = zk_path_topic_tmpl + '{topic}/'

        if self._monitor_unlisted_consumer_groups:
            # don't overwrite self._consumer_groups because that holds the static config values which are always used
            # when fetching consumer offsets from Kafka. Also, these dynamically fetched groups may change on each run.
            consumer_groups = {
                consumer_group: None
                for consumer_group in self._get_zk_path_children(zk_path_consumer, 'consumer groups')
            }
        else:
            consumer_groups = self._consumer_groups

        for consumer_group, topics in consumer_groups.items():
            if not topics:  # If topics are't specified, fetch them from ZK
                zk_path_topics = zk_path_topic_tmpl.format(group=consumer_group)
                topics = {topic: None for topic in self._get_zk_path_children(zk_path_topics, 'topics')}

            for topic, partitions in topics.items():
                if not partitions:  # If partitions aren't specified, fetch them from ZK
                    zk_path_partitions = zk_path_partition_tmpl.format(group=consumer_group, topic=topic)
                    # Zookeeper returns the partition IDs as strings because they are extracted from the node path
                    partitions = [int(x) for x in self._get_zk_path_children(zk_path_partitions, 'partitions')]

                for partition in partitions:
                    zk_path = (zk_path_partition_tmpl + '{partition}/').format(
                        group=consumer_group, topic=topic, partition=partition
                    )
                    try:
                        consumer_offset = int(self._zk_client.get(zk_path)[0])
                        key = (consumer_group, topic, partition)
                        self._zk_consumer_offsets[key] = consumer_offset

                        if len(self._zk_consumer_offsets) >= contexts_limit:
                            self.warning("Context limit reached. Skipping zk consumer offsets collection.")
                            return
                    except NoNodeError:
                        self.log.info('No zookeeper node at %s', zk_path)
                        continue
                    except Exception:
                        self.log.exception('Could not read consumer offset from %s', zk_path)
def _get_zk_consumer_offsets ( self , contexts_limit ) : # Construct the Zookeeper path pattern # /consumers/[groupId]/offsets/[topic]/[partitionId] zk_path_consumer = '/consumers/' zk_path_topic_tmpl = zk_path_consumer + '{group}/offsets/' zk_path_partition_tmpl = zk_path_topic_tmpl + '{topic}/' if self . _monitor_unlisted_consumer_groups : # don't overwrite self._consumer_groups because that holds the static config values which are always used # when fetching consumer offsets from Kafka. Also, these dynamically fetched groups may change on each run. consumer_groups = { consumer_group : None for consumer_group in self . _get_zk_path_children ( zk_path_consumer , 'consumer groups' ) } else : consumer_groups = self . _consumer_groups for consumer_group , topics in consumer_groups . items ( ) : if not topics : # If topics are't specified, fetch them from ZK zk_path_topics = zk_path_topic_tmpl . format ( group = consumer_group ) topics = { topic : None for topic in self . _get_zk_path_children ( zk_path_topics , 'topics' ) } for topic , partitions in topics . items ( ) : if not partitions : # If partitions aren't specified, fetch them from ZK zk_path_partitions = zk_path_partition_tmpl . format ( group = consumer_group , topic = topic ) # Zookeeper returns the partition IDs as strings because they are extracted from the node path partitions = [ int ( x ) for x in self . _get_zk_path_children ( zk_path_partitions , 'partitions' ) ] for partition in partitions : zk_path = ( zk_path_partition_tmpl + '{partition}/' ) . format ( group = consumer_group , topic = topic , partition = partition ) try : consumer_offset = int ( self . _zk_client . get ( zk_path ) [ 0 ] ) key = ( consumer_group , topic , partition ) self . _zk_consumer_offsets [ key ] = consumer_offset if len ( self . _zk_consumer_offsets ) >= contexts_limit : self . warning ( "Context limit reached. Skipping zk consumer offsets collection." ) return except NoNodeError : self . log . info ( 'No zookeeper node at %s' , zk_path ) continue except Exception : self . log . exception ( 'Could not read consumer offset from %s' , zk_path )