docs: def _send_consumer_aware_request(self, group, payloads, encoder_fn, decoder_fn):
          """
          Send a list of requests to the consumer coordinator for the group
          specified using the supplied encode/decode functions. As the payloads
          that use consumer-aware requests do not contain the group (e.g.
          OffsetFetchRequest), all payloads must be for a single group.
  
          Arguments:
  
          group: the name of the consumer group (str) the payloads are for
          payloads: list of object-like entities with topic (str) and
              partition (int) attributes; payloads with duplicate
              topic+partition 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]
  
          broker = self._get_coordinator_for_group(group)
  
          # Send the list of request payloads and collect the responses and
          # errors
          responses = {}
          request_id = self._next_id()
          log.debug('Request %s to %s: %s', request_id, broker, payloads)
          request = encoder_fn(client_id=self.client_id,
                               correlation_id=request_id, payloads=payloads)
  
          # Send the request, recv the response
          try:
              host, port, afi = get_ip_port_afi(broker.host)
              conn = self._get_conn(host, broker.port, afi)
          except KafkaConnectionError as e:
              log.warning('KafkaConnectionError attempting to send request %s '
                          'to server %s: %s', request_id, broker, e)
  
              for payload in payloads:
                  topic_partition = (payload.topic, payload.partition)
                  responses[topic_partition] = FailedPayloadsError(payload)
  
          # No exception, try to get response
          else:
  
              future = conn.send(request_id, request)
              while not future.is_done:
                  for r, f in conn.recv():
                      f.success(r)
  
              # 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)', request_id)
                  for payload in payloads:
                      topic_partition = (payload.topic, payload.partition)
                      responses[topic_partition] = None
                  return []
  
              if future.failed():
                  log.warning('Error attempting to receive a '
                              'response to request %s from server %s: %s',
                              request_id, broker, future.exception)
  
                  for payload in payloads:
                      topic_partition = (payload.topic, payload.partition)
                      responses[topic_partition] = FailedPayloadsError(payload)
  
              else:
                  response = future.value
                  _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', request_id, _resps)
  
          # Return responses in the same order as provided
          return [responses[tp] for tp in original_ordering]