original_docstring: 
          Fetch highwater offsets for each topic/partition from 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. No need to limit this for performance because fetching broker
          offsets from Kafka is a relatively inexpensive operation.
          
docstring: Fetch highwater offsets for each topic/partition from 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. No need to limit this for performance because fetching broker
  offsets from Kafka is a relatively inexpensive operation.
docstring_tokens: Fetch highwater offsets for each topic / partition from 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 . No need to limit this for performance because fetching broker offsets from Kafka is a relatively inexpensive operation .
code: def _get_highwater_offsets(self, kafka_hosts_ports):
          """
          Fetch highwater offsets for each topic/partition from 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. No need to limit this for performance because fetching broker
          offsets from Kafka is a relatively inexpensive operation.
          """
          kafka_conn = SimpleClient(kafka_hosts_ports, timeout=self.kafka_timeout)
          try:
              broker_topics_partitions = kafka_conn.topics_to_brokers.keys()
              # batch a bunch of requests into a single network call
              offsets_request = [OffsetRequestPayload(topic, partition, -1, 1)
                  for topic, partition in broker_topics_partitions]
              offsets_response = kafka_conn.send_offset_request(offsets_request)
              highwater_offsets = {(x.topic, x.partition): x.offsets[0] for x in offsets_response}
          finally:
              try:
                  kafka_conn.close()
              except Exception:
                  self.log.exception('Error cleaning up Kafka connection')
          return highwater_offsets
code_tokens: def _get_highwater_offsets ( self , kafka_hosts_ports ) : kafka_conn = SimpleClient ( kafka_hosts_ports , timeout = self . kafka_timeout ) try : broker_topics_partitions = kafka_conn . topics_to_brokers . keys ( ) offsets_request = [ OffsetRequestPayload ( topic , partition , - 1 , 1 ) for topic , partition in broker_topics_partitions ] offsets_response = kafka_conn . send_offset_request ( offsets_request ) highwater_offsets = { ( x . topic , x . partition ) : x . offsets [ 0 ] for x in offsets_response } finally : try : kafka_conn . close ( ) except Exception : self . log . exception ( 'Error cleaning up Kafka connection' ) return highwater_offsets
short_docstring: Fetch highwater offsets for each topic/partition from Kafka cluster.
short_docstring_tokens: Fetch highwater offsets for each topic / partition from Kafka cluster .
comment: """
          Fetch highwater offsets for each topic/partition from 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. No need to limit this for performance because fetching broker
          offsets from Kafka is a relatively inexpensive operation.
          """$SEP$# batch a bunch of requests into a single network call