docstr: 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
code: 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, wakeup=False)
              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
  
imports: from __future__ import absolute_import
  import collections
  import copy
  import logging
  import random
  import sys
  import time
  from kafka.vendor import six
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.fetch import FetchRequest
  from kafka.protocol.offset import OffsetRequest, OffsetResetStrategy, UNKNOWN_OFFSET
  from kafka.record import MemoryRecords
  from kafka.serializer import Deserializer
  from kafka.structs import TopicPartition, OffsetAndTimestamp
label_desc: Use image node_id 2 for calling the Fetcher obj's underlying member method code with example usage: obj.send_fetches() and returns: futures