def send_fetches(self):
        """Send FetchRequests for all assigned partitions that do not already have
        an in-flight fetch or pending fetch data.

        Returns:
            List of Futures: each future resolves to a FetchResponse
        """
        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.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
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 . 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
Send FetchRequests for all assigned partitions that do not already have
        an in-flight fetch or pending fetch data.

        Returns:
            List of Futures: each future resolves to a FetchResponse