docs: 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]
  
          # Group the requests by topic+partition
          brokers_for_payloads = []
          payloads_by_broker = collections.defaultdict(list)
  
          responses = {}
          for payload in payloads:
              try:
                  leader = self._get_leader_for_partition(payload.topic,
                                                          payload.partition)
                  payloads_by_broker[leader].append(payload)
                  brokers_for_payloads.append(leader)
              except KafkaUnavailableError as e:
                  log.warning('KafkaUnavailableError attempting to send request '
                              'on topic %s partition %d', payload.topic, payload.partition)
                  topic_partition = (payload.topic, payload.partition)
                  responses[topic_partition] = FailedPayloadsError(payload)
  
          # For each broker, send the list of request payloads
          # and collect the responses and errors
          broker_failures = []
  
          # For each KafkaConnection keep the real socket so that we can use
          # a select to perform unblocking I/O
          connections_by_socket = {}
          for broker, payloads in payloads_by_broker.items():
              requestId = self._next_id()
              log.debug('Request %s to %s: %s', requestId, broker, payloads)
              request = encoder_fn(client_id=self.client_id,
                                   correlation_id=requestId, payloads=payloads)
  
              # Send the request, recv the response
              try:
                  conn = self._get_conn(broker.host.decode('utf-8'), broker.port)
                  conn.send(requestId, request)
  
              except ConnectionError as e:
                  broker_failures.append(broker)
                  log.warning('ConnectionError attempting to send request %s '
                              'to server %s: %s', requestId, broker, e)
  
                  for payload in payloads:
                      topic_partition = (payload.topic, payload.partition)
                      responses[topic_partition] = FailedPayloadsError(payload)
  
              # No exception, try to get response
              else:
  
                  # decoder_fn=None signal that the server is expected to not
                  # send a response.  This probably only applies to
                  # ProduceRequest w/ acks = 0
                  if decoder_fn is None:
                      log.debug('Request %s does not expect a response '
                                '(skipping conn.recv)', requestId)
                      for payload in payloads:
                          topic_partition = (payload.topic, payload.partition)
                          responses[topic_partition] = None
                      continue
                  else:
                      connections_by_socket[conn.get_connected_socket()] = (conn, broker, requestId)
  
          conn = None
          while connections_by_socket:
              sockets = connections_by_socket.keys()
              rlist, _, _ = select.select(sockets, [], [], None)
              conn, broker, requestId = connections_by_socket.pop(rlist[0])
              try:
                  response = conn.recv(requestId)
              except ConnectionError as e:
                  broker_failures.append(broker)
                  log.warning('ConnectionError attempting to receive a '
                              'response to request %s from server %s: %s',
                              requestId, broker, e)
  
                  for payload in payloads_by_broker[broker]:
                      topic_partition = (payload.topic, payload.partition)
                      responses[topic_partition] = FailedPayloadsError(payload)
  
              else:
                  _resps = []
                  for payload_response in decoder_fn(response):
                      topic_partition = (payload_response.topic,
                                         payload_response.partition)
                      responses[topic_partition] = payload_response
                      _resps.append(payload_response)
                  log.debug('Response %s: %s', requestId, _resps)
  
          # 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
          if broker_failures:
              self.reset_all_metadata()
  
          # Return responses in the same order as provided
          return [responses[tp] for tp in original_ordering]