docstr: Fetch offset for each partition passed in ``timestamps`` map.
  
  Blocks until offsets are obtained, a non-retriable exception is raised
  or ``timeout_ms`` passed.
  
  Arguments:
      timestamps: {TopicPartition: int} dict with timestamps to fetch
          offsets by. -1 for the latest available, -2 for the earliest
          available. Otherwise timestamp is treated as epoch milliseconds.
  
  Returns:
      {TopicPartition: (int, int)}: Mapping of partition to
          retrieved offset and timestamp. If offset does not exist for
          the provided timestamp, that partition will be missing from
          this mapping.
code: def _retrieve_offsets(self, timestamps, timeout_ms=float("inf")):
      """Fetch offset for each partition passed in ``timestamps`` map.
  
      Blocks until offsets are obtained, a non-retriable exception is raised
      or ``timeout_ms`` passed.
  
      Arguments:
          timestamps: {TopicPartition: int} dict with timestamps to fetch
              offsets by. -1 for the latest available, -2 for the earliest
              available. Otherwise timestamp is treated as epoch milliseconds.
  
      Returns:
          {TopicPartition: (int, int)}: Mapping of partition to
              retrieved offset and timestamp. If offset does not exist for
              the provided timestamp, that partition will be missing from
              this mapping.
      """
      if not timestamps:
          return {}
  
      start_time = time.time()
      remaining_ms = timeout_ms
      timestamps = copy.copy(timestamps)
      while remaining_ms > 0:
          if not timestamps:
              return {}
  
          future = self._send_offset_requests(timestamps)
          self._client.poll(future=future, timeout_ms=remaining_ms)
  
          if future.succeeded():
              return future.value
          if not future.retriable():
              raise future.exception  # pylint: disable-msg=raising-bad-type
  
          elapsed_ms = (time.time() - start_time) * 1000
          remaining_ms = timeout_ms - elapsed_ms
          if remaining_ms < 0:
              break
  
          if future.exception.invalid_metadata:
              refresh_future = self._client.cluster.request_update()
              self._client.poll(
                  future=refresh_future, timeout_ms=remaining_ms
              )
  
              # Issue #1780
              # Recheck partition existence after after a successful metadata refresh
              if refresh_future.succeeded() and isinstance(
                  future.exception, Errors.StaleMetadata
              ):
                  log.debug(
                      "Stale metadata was raised, and we now have an updated metadata. Rechecking partition existence"
                  )
                  unknown_partition = future.exception.args[
                      0
                  ]  # TopicPartition from StaleMetadata
                  if (
                      self._client.cluster.leader_for_partition(
                          unknown_partition
                      )
                      is None
                  ):
                      log.debug(
                          "Removed partition %s from offsets retrieval"
                          % (unknown_partition,)
                      )
                      timestamps.pop(unknown_partition)
          else:
              time.sleep(self.config["retry_backoff_ms"] / 1000.0)
  
          elapsed_ms = (time.time() - start_time) * 1000
          remaining_ms = timeout_ms - elapsed_ms
  
      raise Errors.KafkaTimeoutError(
          "Failed to get offsets by timestamps in %s ms" % (timeout_ms,)
      )
  
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 12 for calling the Fetcher obj's underlying member method code with example usage: obj._retrieve_offsets(timestamps, timeout_ms) and returns: dict, dict, future