@Test
    public void testRemoveAndMarkSnapshotForDeletionAlreadyDeleted() throws IOException {
        File file = LogFileUtils.producerSnapshotFile(logDir, 5);
        Files.createFile(file.toPath());
        ProducerStateManager manager = new ProducerStateManager(partition, logDir, maxTransactionTimeoutMs, producerStateManagerConfig, time);
        assertTrue(manager.latestSnapshotOffset().isPresent());
        Files.delete(file.toPath());
        assertFalse(manager.removeAndMarkSnapshotForDeletion(5).isPresent());
        assertFalse(manager.latestSnapshotOffset().isPresent());
    }
Test that marking a snapshot for deletion when the file has already been deleted
returns None instead of the SnapshotFile. The snapshot file should be removed from
the in-memory state of the ProducerStateManager. This scenario can occur during log
recovery when the intermediate ProducerStateManager instance deletes a file without
updating the state of the "real" ProducerStateManager instance which is passed to the Log.
storage/src/test/java/org/apache/kafka/storage/internals/log/ProducerStateManagerTest.java