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]
  
          # 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]