#NEWLINE# def seek( self , partition , offset ) : #NEWLINE# #INDENT# 'Manually specify the fetch offset for a TopicPartition.\n\n        Overrides the fetch offsets that the consumer will use on the next\n        poll(). If this API is invoked for the same partition more than once,\n        the latest offset will be used on the next poll(). Note that you may\n        lose data if this API is arbitrarily used in the middle of consumption,\n        to reset the fetch offsets.\n\n        Arguments:\n            partition (TopicPartition): partition for seek operation\n            offset (int): message offset in partition\n\n        Raises:\n            AssertionError: if offset is not an int >= 0; or if partition is not\n                currently assigned.\n        ' #NEWLINE# if ( not isinstance ( partition , TopicPartition ) ) : #NEWLINE# #INDENT# raise TypeError ( 'partition must be a TopicPartition namedtuple' ) #NEWLINE# #UNINDENT# assert ( isinstance ( offset , int ) and ( offset >= 0 ) ) , 'Offset must be >= 0' #NEWLINE# assert ( partition in self . _subscription . assigned_partitions ( ) ) , 'Unassigned partition' #NEWLINE# log . debug ( 'Seeking to offset %s for partition %s' , offset , partition ) #NEWLINE# self . _subscription . assignment [ partition ] . seek ( offset )