'Encode and send a FetchRequest
Payloads are grouped by topic and partition so they can be pipelined
to the same brokers.'

def send_fetch_request(self, payloads=[], fail_on_error=True, callback=None, max_wait_time=100, min_bytes=4096):

    encoder = functools.partial(KafkaProtocol.encode_fetch_request, max_wait_time=max_wait_time, min_bytes=min_bytes)
    resps = self._send_broker_aware_request(payloads, encoder, KafkaProtocol.decode_fetch_response)
    return [(resp if (not callback) else callback(resp)) for resp in resps if ((not fail_on_error) or (not self._raise_on_response_error(resp)))]
