'Send FetchRequests asynchronously for all assigned partitions.
Note: noop if there are unconsumed records internal to the fetcher
Returns:
List of Futures: each future resolves to a FetchResponse'

def send_fetches(self):

    futures = []
    for (node_id, request) in six.iteritems(self._create_fetch_requests()):
        if self._client.ready(node_id):
            log.debug('Sending    FetchRequest    to    node    %s', node_id)
            future = self._client.send(node_id, request)
            future.error_on_callbacks = True
            future.add_callback(self._handle_fetch_response, request, time.time())
            future.add_errback(log.error, 'Fetch    to    node    %s    failed:    %s', node_id)
            futures.append(future)
    self._fetch_futures.extend(futures)
    self._clean_done_fetch_futures()
    return futures
