docstr: Fetch offsets for each partition in timestamps dict. This may send
  request to multiple nodes, based on who is Leader for partition.
  
  Arguments:
      timestamps (dict): {TopicPartition: int} mapping of fetching
          timestamps.
  
  Returns:
      Future: resolves to a mapping of retrieved offsets
code: def _send_offset_requests(self, timestamps):
      """Fetch offsets for each partition in timestamps dict. This may send
      request to multiple nodes, based on who is Leader for partition.
  
      Arguments:
          timestamps (dict): {TopicPartition: int} mapping of fetching
              timestamps.
  
      Returns:
          Future: resolves to a mapping of retrieved offsets
      """
      timestamps_by_node = collections.defaultdict(dict)
      for partition, timestamp in six.iteritems(timestamps):
          node_id = self._client.cluster.leader_for_partition(partition)
          if node_id is None:
              self._client.add_topic(partition.topic)
              log.debug(
                  "Partition %s is unknown for fetching offset,"
                  " wait for metadata refresh",
                  partition,
              )
              return Future().failure(Errors.StaleMetadata(partition))
          elif node_id == -1:
              log.debug(
                  "Leader for partition %s unavailable for fetching "
                  "offset, wait for metadata refresh",
                  partition,
              )
              return Future().failure(
                  Errors.LeaderNotAvailableError(partition)
              )
          else:
              timestamps_by_node[node_id][partition] = timestamp
  
      # Aggregate results until we have all
      list_offsets_future = Future()
      responses = []
      node_count = len(timestamps_by_node)
  
      def on_success(value):
          responses.append(value)
          if len(responses) == node_count:
              offsets = {}
              for r in responses:
                  offsets.update(r)
              list_offsets_future.success(offsets)
  
      def on_fail(err):
          if not list_offsets_future.is_done:
              list_offsets_future.failure(err)
  
      for node_id, timestamps in six.iteritems(timestamps_by_node):
          _f = self._send_offset_request(node_id, timestamps)
          _f.add_callback(on_success)
          _f.add_errback(on_fail)
      return list_offsets_future
  
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 20 for calling the Fetcher obj's underlying member method code with example usage: obj._send_offset_requests(timestamps) and returns: list_offsets_future, Future, Future