def _send_consumer_aware_request(self, group, payloads, encoder_fn, decoder_fn):
<EOL>original_ordering = [(p.topic, p.partition) for p in payloads]<EOL>broker = self._get_coordinator_for_group(group)<EOL>responses = {}<EOL>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>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>return []<EOL><DEDENT>try:<EOL><INDENT>response = conn.recv(requestId)<EOL><DEDENT>except ConnectionError as e:<EOL><INDENT>log.warning('<STR_LIT>'<EOL>'<STR_LIT>',<EOL>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>_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>return [responses[tp] for tp in original_ordering]<EOL>
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