docs: 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 miliseconds.
  
          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
          while remaining_ms > 0:
              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)
              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,))