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