def _send_broker_aware_request(self, payloads, encoder_fn, decoder_fn):
<EOL>original_ordering = [(p.topic, p.partition) for p in payloads]<EOL>brokers_for_payloads = []<EOL>payloads_by_broker = collections.defaultdict(list)<EOL>responses = {}<EOL>for payload in payloads:<EOL><INDENT>try:<EOL><INDENT>leader = self._get_leader_for_partition(payload.topic,<EOL>payload.partition)<EOL>payloads_by_broker[leader].append(payload)<EOL>brokers_for_payloads.append(leader)<EOL><DEDENT>except KafkaUnavailableError as e:<EOL><INDENT>log.warning('<STR_LIT>'<EOL>'<STR_LIT>', payload.topic, payload.partition)<EOL>topic_partition = (payload.topic, payload.partition)<EOL>responses[topic_partition] = FailedPayloadsError(payload)<EOL><DEDENT><DEDENT>broker_failures = []<EOL>connections_by_socket = {}<EOL>for broker, payloads in payloads_by_broker.items():<EOL><INDENT>requestId = self._next_id()<EOL>log.debug('<STR_LIT>', requestId, broker, payloads)<EOL>request = encoder_fn(client_id=self.client_id,<EOL>correlation_id=requestId, payloads=payloads)<EOL>try:<EOL><INDENT>conn = self._get_conn(broker.host.decode('<STR_LIT:utf-8>'), broker.port)<EOL>conn.send(requestId, request)<EOL><DEDENT>except ConnectionError as e:<EOL><INDENT>broker_failures.append(broker)<EOL>log.warning('<STR_LIT>'<EOL>'<STR_LIT>', requestId, broker, e)<EOL>for payload in payloads:<EOL><INDENT>topic_partition = (payload.topic, payload.partition)<EOL>responses[topic_partition] = FailedPayloadsError(payload)<EOL><DEDENT><DEDENT>else:<EOL><INDENT>if decoder_fn is None:<EOL><INDENT>log.debug('<STR_LIT>'<EOL>'<STR_LIT>', requestId)<EOL>for payload in payloads:<EOL><INDENT>topic_partition = (payload.topic, payload.partition)<EOL>responses[topic_partition] = None<EOL><DEDENT>continue<EOL><DEDENT>else:<EOL><INDENT>connections_by_socket[conn.get_connected_socket()] = (conn, broker, requestId)<EOL><DEDENT><DEDENT><DEDENT>conn = None<EOL>while connections_by_socket:<EOL><INDENT>sockets = connections_by_socket.keys()<EOL>rlist, _, _ = select.select(sockets, [], [], None)<EOL>conn, broker, requestId = connections_by_socket.pop(rlist[<NUM_LIT:0>])<EOL>try:<EOL><INDENT>response = conn.recv(requestId)<EOL><DEDENT>except ConnectionError as e:<EOL><INDENT>broker_failures.append(broker)<EOL>log.warning('<STR_LIT>'<EOL>'<STR_LIT>',<EOL>requestId, broker, e)<EOL>for payload in payloads_by_broker[broker]:<EOL><INDENT>topic_partition = (payload.topic, payload.partition)<EOL>responses[topic_partition] = FailedPayloadsError(payload)<EOL><DEDENT><DEDENT>else:<EOL><INDENT>_resps = []<EOL>for payload_response in decoder_fn(response):<EOL><INDENT>topic_partition = (payload_response.topic,<EOL>payload_response.partition)<EOL>responses[topic_partition] = payload_response<EOL>_resps.append(payload_response)<EOL><DEDENT>log.debug('<STR_LIT>', requestId, _resps)<EOL><DEDENT><DEDENT>if broker_failures:<EOL><INDENT>self.reset_all_metadata()<EOL><DEDENT>return [responses[tp] for tp in original_ordering]<EOL>
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