(self, response)
Callback that parses an OffsetFetchResponse and saves it to the highwater_offsets dict.
Callback that parses an OffsetFetchResponse and saves it to the highwater_offsets dict.
Callback that parses an OffsetFetchResponse and saves it to the highwater_offsets dict .
def _highwater_offsets_callback(self, response):
        """Callback that parses an OffsetFetchResponse and saves it to the highwater_offsets dict."""
        if type(response) not in OffsetResponse:
            raise RuntimeError("response type should be OffsetResponse, but instead was %s." % type(response))
        for topic, partitions_data in response.topics:
            for partition, error_code, offsets in partitions_data:
                error_type = kafka_errors.for_code(error_code)
                if error_type is kafka_errors.NoError:
                    self._highwater_offsets[(topic, partition)] = offsets[0]
                elif error_type is kafka_errors.NotLeaderForPartitionError:
                    self.log.warning(
                        "Kafka broker returned %s (error_code %s) for topic %s, partition: %s. This should only happen "
                        "if the broker that was the partition leader when kafka_admin_client last fetched metadata is "
                        "no longer the leader.",
                        error_type.message,
                        error_type.errno,
                        topic,
                        partition,
                    )
                    self.kafka_client._client.cluster.request_update()  # force metadata update on next poll()
                elif error_type is kafka_errors.UnknownTopicOrPartitionError:
                    self.log.warning(
                        "Kafka broker returned %s (error_code %s) for topic: %s, partition: %s. This should only "
                        "happen if the topic is currently being deleted or the check configuration lists non-existent "
                        "topic partitions.",
                        error_type.message,
                        error_type.errno,
                        topic,
                        partition,
                    )
                else:
                    raise error_type(
                        "Unexpected error encountered while attempting to fetch the highwater offsets for topic: %s, "
                        "partition: %s." % (topic, partition)
                    )
def _highwater_offsets_callback ( self , response ) : if type ( response ) not in OffsetResponse : raise RuntimeError ( "response type should be OffsetResponse, but instead was %s." % type ( response ) ) for topic , partitions_data in response . topics : for partition , error_code , offsets in partitions_data : error_type = kafka_errors . for_code ( error_code ) if error_type is kafka_errors . NoError : self . _highwater_offsets [ ( topic , partition ) ] = offsets [ 0 ] elif error_type is kafka_errors . NotLeaderForPartitionError : self . log . warning ( "Kafka broker returned %s (error_code %s) for topic %s, partition: %s. This should only happen " "if the broker that was the partition leader when kafka_admin_client last fetched metadata is " "no longer the leader." , error_type . message , error_type . errno , topic , partition , ) self . kafka_client . _client . cluster . request_update ( ) # force metadata update on next poll() elif error_type is kafka_errors . UnknownTopicOrPartitionError : self . log . warning ( "Kafka broker returned %s (error_code %s) for topic: %s, partition: %s. This should only " "happen if the topic is currently being deleted or the check configuration lists non-existent " "topic partitions." , error_type . message , error_type . errno , topic , partition , ) else : raise error_type ( "Unexpected error encountered while attempting to fetch the highwater offsets for topic: %s, " "partition: %s." % ( topic , partition ) )