public Optional<SnapshotFile> removeAndMarkSnapshotForDeletion(long snapshotOffset) throws IOException {
        SnapshotFile snapshotFile = snapshots.remove(snapshotOffset);
        if (snapshotFile != null) {
            // If the file cannot be renamed, it likely means that the file was deleted already.
            // This can happen due to the way we construct an intermediate producer state manager
            // during log recovery, and use it to issue deletions prior to creating the "real"
            // producer state manager.
            //
            // In any case, removeAndMarkSnapshotForDeletion is intended to be used for snapshot file
            // deletion, so ignoring the exception here just means that the intended operation was
            // already completed.
            try {
                snapshotFile.renameToDelete();
                return Optional.of(snapshotFile);
            } catch (NoSuchFileException ex) {
                log.info("Failed to rename producer state snapshot {} with deletion suffix because it was already deleted", snapshotFile.file().getAbsoluteFile());
            }
        }
        return Optional.empty();
    }
Removes the producer state snapshot file metadata corresponding to the provided offset if it exists from this
ProducerStateManager, and renames the backing snapshot file to have the Log.DeletionSuffix.
<p>
Note: This method is safe to use with async deletes. If a race occurs and the snapshot file
is deleted without this ProducerStateManager instance knowing, the resulting exception on
SnapshotFile rename will be ignored and {@link Optional#empty()} will be returned.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java