(self, contexts_limit)
Fetch highwater offsets for topic_partitions in the Kafka cluster.

        If monitor_all_broker_highwatermarks is True, will fetch for all partitions in the cluster. Otherwise highwater
        mark offsets will only be fetched for topic partitions where this check run has already fetched a consumer
        offset.

        Internal Kafka topics like __consumer_offsets, __transaction_state, etc are always excluded.

        Any partitions that don't currently have a leader will be skipped.

        Sends one OffsetRequest per broker to get offsets for all partitions where that broker is the leader:
        https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetAPI(AKAListOffset)
Fetch highwater offsets for topic_partitions in the Kafka cluster.
Fetch highwater offsets for topic_partitions in the Kafka cluster .
def _get_highwater_offsets(self, contexts_limit):
        """
        Fetch highwater offsets for topic_partitions in the Kafka cluster.

        If monitor_all_broker_highwatermarks is True, will fetch for all partitions in the cluster. Otherwise highwater
        mark offsets will only be fetched for topic partitions where this check run has already fetched a consumer
        offset.

        Internal Kafka topics like __consumer_offsets, __transaction_state, etc are always excluded.

        Any partitions that don't currently have a leader will be skipped.

        Sends one OffsetRequest per broker to get offsets for all partitions where that broker is the leader:
        https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetAPI(AKAListOffset)
        """
        # If we aren't fetching all broker highwater offsets, then construct the unique set of topic partitions for
        # which this run of the check has at least once saved consumer offset. This is later used as a filter for
        # excluding partitions.
        if not self._monitor_all_broker_highwatermarks:
            tps_with_consumer_offset = {(topic, partition) for (_, topic, partition) in self._kafka_consumer_offsets}
            tps_with_consumer_offset.update({(topic, partition) for (_, topic, partition) in self._zk_consumer_offsets})

        for broker in self.kafka_client.cluster.brokers():
            if len(self._highwater_offsets) >= contexts_limit:
                self.warning("Context limit reached. Skipping highwater offsets collection.")
                return
            broker_led_partitions = self.kafka_client.cluster.partitions_for_broker(broker.nodeId)
            if broker_led_partitions is None:
                continue
            # Take the partitions for which this broker is the leader and group them by topic in order to construct the
            # OffsetRequest while simultaneously filtering out partitions we want to exclude
            partitions_grouped_by_topic = defaultdict(list)
            for topic, partition in broker_led_partitions:
                # No sense fetching highwater offsets for internal topics
                if topic not in KAFKA_INTERNAL_TOPICS and (
                    self._monitor_all_broker_highwatermarks or (topic, partition) in tps_with_consumer_offset
                ):
                    partitions_grouped_by_topic[topic].append(partition)

            # Construct the OffsetRequest
            max_offsets = 1
            request = OffsetRequest[0](
                replica_id=-1,
                topics=[
                    (topic, [(partition, OffsetResetStrategy.LATEST, max_offsets) for partition in partitions])
                    for topic, partitions in partitions_grouped_by_topic.items()
                ],
            )
            response = self._make_blocking_req(request, node_id=broker.nodeId)
            self._process_highwater_offsets(response, contexts_limit)
def _get_highwater_offsets ( self , contexts_limit ) : # If we aren't fetching all broker highwater offsets, then construct the unique set of topic partitions for # which this run of the check has at least once saved consumer offset. This is later used as a filter for # excluding partitions. if not self . _monitor_all_broker_highwatermarks : tps_with_consumer_offset = { ( topic , partition ) for ( _ , topic , partition ) in self . _kafka_consumer_offsets } tps_with_consumer_offset . update ( { ( topic , partition ) for ( _ , topic , partition ) in self . _zk_consumer_offsets } ) for broker in self . kafka_client . cluster . brokers ( ) : if len ( self . _highwater_offsets ) >= contexts_limit : self . warning ( "Context limit reached. Skipping highwater offsets collection." ) return broker_led_partitions = self . kafka_client . cluster . partitions_for_broker ( broker . nodeId ) if broker_led_partitions is None : continue # Take the partitions for which this broker is the leader and group them by topic in order to construct the # OffsetRequest while simultaneously filtering out partitions we want to exclude partitions_grouped_by_topic = defaultdict ( list ) for topic , partition in broker_led_partitions : # No sense fetching highwater offsets for internal topics if topic not in KAFKA_INTERNAL_TOPICS and ( self . _monitor_all_broker_highwatermarks or ( topic , partition ) in tps_with_consumer_offset ) : partitions_grouped_by_topic [ topic ] . append ( partition ) # Construct the OffsetRequest max_offsets = 1 request = OffsetRequest [ 0 ] ( replica_id = - 1 , topics = [ ( topic , [ ( partition , OffsetResetStrategy . LATEST , max_offsets ) for partition in partitions ] ) for topic , partitions in partitions_grouped_by_topic . items ( ) ] , ) response = self . _make_blocking_req ( request , node_id = broker . nodeId ) self . _process_highwater_offsets ( response , contexts_limit )