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):
        """
        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]