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

        Do this for all partitions in the cluster because even if it has no consumers, we may want to measure whether
        producers are successfully producing.

        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)

        For speed, all the brokers are queried in parallel using callbacks. The callback flow is:
            1. Issue an OffsetRequest to every broker
            2. Attach a callback to each OffsetResponse that parses the response and saves the highwater offsets.
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):
        """Fetch highwater offsets for topic_partitions in the Kafka cluster.

        Do this for all partitions in the cluster because even if it has no consumers, we may want to measure whether
        producers are successfully producing.

        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)

        For speed, all the brokers are queried in parallel using callbacks. The callback flow is:
            1. Issue an OffsetRequest to every broker
            2. Attach a callback to each OffsetResponse that parses the response and saves the highwater offsets.
        """
        highwater_futures = []  # No need to store on object because the callbacks don't create additional futures

        # 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._consumer_offsets}

        for batch in self.batchify(self.kafka_client._client.cluster.brokers(), self._broker_requests_batch_size):
            for broker in batch:
                broker_led_partitions = self.kafka_client._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()
                    ],
                )

                highwater_future = self.kafka_client._send_request_to_node(node_id=broker.nodeId, request=request)
                highwater_future.add_callback(self._highwater_offsets_callback)
                highwater_futures.append(highwater_future)

            # Loop until all futures resolved.
            self.kafka_client._wait_for_futures(highwater_futures)
def _get_highwater_offsets ( self ) : highwater_futures = [ ] # No need to store on object because the callbacks don't create additional futures # 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 . _consumer_offsets } for batch in self . batchify ( self . kafka_client . _client . cluster . brokers ( ) , self . _broker_requests_batch_size ) : for broker in batch : broker_led_partitions = self . kafka_client . _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 ( ) ] , ) highwater_future = self . kafka_client . _send_request_to_node ( node_id = broker . nodeId , request = request ) highwater_future . add_callback ( self . _highwater_offsets_callback ) highwater_futures . append ( highwater_future ) # Loop until all futures resolved. self . kafka_client . _wait_for_futures ( highwater_futures )