docs: 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