def _send_broker_aware_request(self, payloads, encoder_fn, decoder_fn):
        """
        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
        """
        # encoders / decoders do not maintain ordering currently
        # so we need to keep this so we can rebuild order before returning
        original_ordering = [(p.topic, p.partition) for p in payloads]

        # Connection errors generally mean stale metadata
        # although sometimes it means incorrect api request
        # Unfortunately there is no good way to tell the difference
        # so we'll just reset metadata on all errors to be safe
        refresh_metadata = False

        # For each broker, send the list of request payloads
        # and collect the responses and errors
        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)

        # For each BrokerConnection keep the real socket so that we can use
        # a select to perform unblocking I/O
        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 KafkaConnectionError:
                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())

            # block until a socket is ready to be read
            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]
                    for r, f in conn.recv():
                        f.success(r)
                    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 in the same order as provided
        return [responses[tp] for tp in original_ordering]
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