#NEWLINE# def _reset_offset( self , partition ) : #NEWLINE# #INDENT# 'Reset offsets for the given partition using the offset reset strategy.\n\n        Arguments:\n            partition (TopicPartition): the partition that needs reset offset\n\n        Raises:\n            NoOffsetForPartitionError: if no offset reset strategy is defined\n        ' #NEWLINE# timestamp = self . _subscriptions . assignment [ partition ] . reset_strategy #NEWLINE# if ( timestamp is OffsetResetStrategy . EARLIEST ) : #NEWLINE# #INDENT# strategy = 'earliest' #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( timestamp is OffsetResetStrategy . LATEST ) : #NEWLINE# #INDENT# strategy = 'latest' #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# raise NoOffsetForPartitionError ( partition ) #NEWLINE# #UNINDENT# #UNINDENT# log . debug ( 'Resetting offset for partition %s to %s offset.' , partition , strategy ) #NEWLINE# offset = self . _offset ( partition , timestamp ) #NEWLINE# if self . _subscriptions . is_assigned ( timestamp ) : #NEWLINE# #INDENT# self . _subscriptions . seek ( partition , offset )