docs: def _get_broker_offsets(self, instance, topics):
          """
          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.
  
          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)
  
          Can we cleanup connections on agent restart?
          Brokers before 0.9 - accumulate stale connections on restarts.
          In 0.9 Kafka added connections.max.idle.ms
          https://issues.apache.org/jira/browse/KAFKA-1282
          """
  
          # Connect to Kafka
          highwater_offsets = {}
          topic_partitions_without_a_leader = []
          topics_to_fetch = defaultdict(set)
          cli = self._get_kafka_client(instance)
  
          for topic, partitions in iteritems(topics):
              # if no partitions are provided
              # we're falling back to all available partitions (?)
              if len(partitions) == 0:
                  partitions = cli.cluster.available_partitions_for_topic(topic)
              topics_to_fetch[topic].update(partitions)
  
          leader_tp = defaultdict(lambda: defaultdict(set))
          for topic, partitions in iteritems(topics_to_fetch):
              for partition in partitions:
                  partition_leader = cli.cluster.leader_for_partition(TopicPartition(topic, partition))
                  if partition_leader is not None and partition_leader >= 0:
                      leader_tp[partition_leader][topic].add(partition)
  
          max_offsets = 1
          for node_id, tps in iteritems(leader_tp):
              # Construct the OffsetRequest
              request = OffsetRequest[0](
                  replica_id=-1,
                  topics=[
                      (topic, [(partition, OffsetResetStrategy.LATEST, max_offsets) for partition in partitions])
                      for topic, partitions in iteritems(tps)
                  ],
              )
  
              response = self._make_blocking_req(cli, request, node_id=node_id)
              offsets, unled = self._process_highwater_offsets(response)
              highwater_offsets.update(offsets)
              topic_partitions_without_a_leader.extend(unled)
  
          return highwater_offsets, list(set(topic_partitions_without_a_leader))