@Test
    public void testEntryForVerification() {
        VerificationStateEntry originalEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, epoch);
        VerificationGuard originalEntryVerificationGuard = originalEntry.verificationGuard();

        // If we already have an entry, reuse it.
        VerificationStateEntry updateEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, epoch);
        VerificationStateEntry entry = stateManager.verificationStateEntry(producerId);
        assertEquals(originalEntryVerificationGuard, entry.verificationGuard());
        assertEquals(entry.verificationGuard(), updateEntry.verificationGuard());


        // Add the transactional data and clear the entry.
        appendClientEntry(stateManager, producerId, (short) 0, defaultSequence, 0, true);
        stateManager.clearVerificationStateEntry(producerId);
        assertNull(stateManager.verificationStateEntry(producerId));
    }
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