'Manually specify the fetch offset for a TopicPartition.
Overrides the fetch offsets that the consumer will use on the next
:meth:`~kafka.KafkaConsumer.poll`. If this API is invoked for the same
partition more than once, the latest offset will be used on the next
:meth:`~kafka.KafkaConsumer.poll`.
Note: You may lose data if this API is arbitrarily used in the middle of
consumption to reset the fetch offsets.
Arguments:
partition (TopicPartition): Partition for seek operation
offset (int): Message offset in partition
Raises:
AssertionError: If offset is not an int >= 0; or if partition is not
currently assigned.'

def seek(self, partition, offset):

    if (not isinstance(partition, TopicPartition)):
        raise TypeError('partition    must    be    a    TopicPartition    namedtuple')
    assert (isinstance(offset, int) and (offset >= 0)), 'Offset    must    be    >=    0'
    assert (partition in self._subscription.assigned_partitions()), 'Unassigned    partition'
    log.debug('Seeking    to    offset    %s    for    partition    %s', offset, partition)
    self._subscription.assignment[partition].seek(offset)
