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

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))