code: def offsets_for_times(self, timestamps):
          
          if self.config['api_version'] <= (0, 10, 0):
              raise UnsupportedVersionError(
                  "offsets_for_times API not supported for cluster version {}"
                  .format(self.config['api_version']))
          for tp, ts in six.iteritems(timestamps):
              timestamps[tp] = int(ts)
              if ts < 0:
                  raise ValueError(
                      "The target time for partition {} is {}. The target time "
                      "cannot be negative.".format(tp, ts))
          return self._fetcher.get_offsets_by_times(
              timestamps, self.config['request_timeout_ms'])
docs: Look up the offsets for the given partitions by timestamp. The
          returned offset for each partition is the earliest offset whose
          timestamp is greater than or equal to the given timestamp in the
          corresponding partition.
  
          This is a blocking call. The consumer does not have to be assigned the
          partitions.
  
          If the message format version in a partition is before 0.10.0, i.e.
          the messages do not have timestamps, ``None`` will be returned for that
          partition. ``None`` will also be returned for the partition if there
          are no messages in it.
  
          Note:
              This method may block indefinitely if the partition does not exist.
  
          Arguments:
              timestamps (dict): ``{TopicPartition: int}`` mapping from partition
                  to the timestamp to look up. Unit should be milliseconds since
                  beginning of the epoch (midnight Jan 1, 1970 (UTC))
  
          Returns:
              ``{TopicPartition: OffsetAndTimestamp}``: mapping from partition
              to the timestamp and offset of the first message with timestamp
              greater than or equal to the target timestamp.
  
          Raises:
              ValueError: If the target timestamp is negative
              UnsupportedVersionError: If the broker does not support looking
                  up the offsets by timestamp.
              KafkaTimeoutError: If fetch failed in request_timeout_ms