'Reset offsets for the given partition using the offset reset strategy.
Arguments:
partition (TopicPartition): the partition that needs reset offset
Raises:
NoOffsetForPartitionError: if no offset reset strategy is defined'

def _reset_offset(self, partition):

    timestamp = self._subscriptions.assignment[partition].reset_strategy
    if (timestamp is OffsetResetStrategy.EARLIEST):
        strategy = 'earliest'
    elif (timestamp is OffsetResetStrategy.LATEST):
        strategy = 'latest'
    else:
        raise NoOffsetForPartitionError(partition)
    log.debug('Resetting    offset    for    partition    %s    to    %s    offset.', partition, strategy)
    offsets = self._retrieve_offsets({partition: timestamp})
    if (partition not in offsets):
        raise NoOffsetForPartitionError(partition)
    offset = offsets[partition][0]
    if self._subscriptions.is_assigned(partition):
        self._subscriptions.seek(partition, offset)
