'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'

def _send_consumer_aware_request(self, group, payloads, encoder_fn, decoder_fn):

    original_ordering = [(p.topic, p.partition) for p in payloads]
    broker = self._get_coordinator_for_group(group)
    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)
    try:
        (host, port, afi) = get_ip_port_afi(broker.host)
        conn = self._get_conn(host, broker.port, afi)
        conn.send(request_id, request)
    except ConnectionError as e:
        log.warning('ConnectionError    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)
    else:
        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 []
        try:
            response = conn.recv(request_id)
        except ConnectionError as e:
            log.warning('ConnectionError    attempting    to    receive    a    response    to    request    %s    from    server    %s:    %s', request_id, broker, e)
            for payload in payloads:
                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', request_id, _resps)
    return [responses[tp] for tp in original_ordering]
