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, fetch_offsets) in six.iteritems(self._create_fetch_requests()):
            log.debug("Sending FetchRequest to node %s", node_id)
            self._nodes_with_pending_fetch_requests.add(node_id)
            future = self._client.send(node_id, request, wakeup=False)
            future.add_callback(self._handle_fetch_response, node_id, fetch_offsets, time.time())
            future.add_errback(self._handle_fetch_error, node_id)
            future.add_both(self._clear_pending_fetch_request, 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
        
kafka/consumer/fetcher.py