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();
    }
@Test
    public void testRemoveAndMarkSnapshotForDeletion() throws IOException {
        Files.createFile(LogFileUtils.producerSnapshotFile(logDir, 5).toPath());
        ProducerStateManager manager = new ProducerStateManager(partition, logDir, maxTransactionTimeoutMs, producerStateManagerConfig, time);
        assertTrue(manager.latestSnapshotOffset().isPresent());
        Optional<SnapshotFile> snapshotFileOpt = manager.removeAndMarkSnapshotForDeletion(5);
        assertTrue(snapshotFileOpt.isPresent());
        assertTrue(snapshotFileOpt.get().file().toPath().toString().endsWith(LogFileUtils.DELETED_FILE_SUFFIX));
        assertFalse(manager.latestSnapshotOffset().isPresent());
    }