private void testLoadFromCorruptSnapshot(Consumer<FileChannel> consumer) throws IOException {
        appendClientEntry(stateManager, producerId, epoch, 0, 0L, false);
        stateManager.takeSnapshot();

        appendClientEntry(stateManager, producerId, epoch, 1, 1L, false);
        stateManager.takeSnapshot();

        // Truncate the last snapshot
        OptionalLong latestSnapshotOffsetOpt = stateManager.latestSnapshotOffset();
        assertTrue(latestSnapshotOffsetOpt.isPresent());
        long latestSnapshotOffset = latestSnapshotOffsetOpt.getAsLong();
        assertEquals(2L, latestSnapshotOffset);
        File snapshotToTruncate = LogFileUtils.producerSnapshotFile(logDir, latestSnapshotOffset);
        try (FileChannel channel = FileChannel.open(snapshotToTruncate.toPath(), StandardOpenOption.WRITE)) {
            consumer.accept(channel);
        }

        // Ensure that the truncated snapshot is deleted and producer state is loaded from the previous snapshot
        ProducerStateManager reloadedStateManager = new ProducerStateManager(partition, logDir,
                maxTransactionTimeoutMs, producerStateManagerConfig, time);
        reloadedStateManager.truncateAndReload(0L, 20L, time.milliseconds());
        assertFalse(snapshotToTruncate.exists());

        ProducerStateEntry loadedProducerState = reloadedStateManager.activeProducers().get(producerId);
        assertNotNull(loadedProducerState);
        assertEquals(0L, loadedProducerState.lastDataOffset());
    }
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