'Group a list of request payloads by topic+partition and send them to
the leader broker for that partition using the supplied encode/decode
functions
Arguments:
payloads: list of object-like entities with a topic (str) and
partition (int) attribute; payloads with duplicate topic-partitions
are not supported.
encode_fn: a method to encode the list of payloads to a request body,
must accept client_id, correlation_id, and payloads as
keyword arguments
decode_fn: a method to decode a response body into response objects.
The response objects must be object-like and have topic
and partition attributes
Returns:
List of response objects in the same order as the supplied payloads'

def _send_broker_aware_request(self, payloads, encoder_fn, decoder_fn):

    original_ordering = [(p.topic, p.partition) for p in payloads]
    refresh_metadata = False
    payloads_by_broker = self._payloads_by_broker(payloads)
    responses = {}
    def failed_payloads(payloads):
        for payload in payloads:
            topic_partition = (str(payload.topic), payload.partition)
            responses[topic_partition] = FailedPayloadsError(payload)
    connections_by_future = {}
    for (broker, broker_payloads) in six.iteritems(payloads_by_broker):
        if (broker is None):
            failed_payloads(broker_payloads)
            continue
        (host, port, afi) = get_ip_port_afi(broker.host)
        try:
            conn = self._get_conn(host, broker.port, afi)
        except ConnectionError:
            refresh_metadata = True
            failed_payloads(broker_payloads)
            continue
        request = encoder_fn(payloads=broker_payloads)
        future = conn.send(request)
        if future.failed():
            refresh_metadata = True
            failed_payloads(broker_payloads)
            continue
        if (not request.expect_response()):
            for payload in broker_payloads:
                topic_partition = (str(payload.topic), payload.partition)
                responses[topic_partition] = None
            continue
        connections_by_future[future] = (conn, broker)
    conn = None
    while connections_by_future:
        futures = list(connections_by_future.keys())
        sockets = [conn._sock for (future, (conn, _)) in six.iteritems(connections_by_future) if ((not future.is_done) and (conn._sock is not None))]
        if sockets:
            (read_socks, _, _) = select.select(sockets, [], [])
        for future in futures:
            if (not future.is_done):
                (conn, _) = connections_by_future[future]
                conn.recv()
                continue
            (_, broker) = connections_by_future.pop(future)
            if future.failed():
                refresh_metadata = True
                failed_payloads(payloads_by_broker[broker])
            else:
                for payload_response in decoder_fn(future.value):
                    topic_partition = (str(payload_response.topic), payload_response.partition)
                    responses[topic_partition] = payload_response
    if refresh_metadata:
        self.reset_all_metadata()
    return [responses[tp] for tp in original_ordering]
