'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'

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 timestamps.items():
        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'])
