@Test
    public void testSkipEmptyTransactions() {
        AtomicInteger seq = new AtomicInteger(defaultSequence);

        // Start one transaction in a separate append
        ProducerAppendInfo firstAppend = stateManager.prepareUpdate(producerId, AppendOrigin.CLIENT);
        appendData(16L, 20L, firstAppend, seq);
        assertTxnMetadataEquals(new TxnMetadata(producerId, 16L), firstAppend.startedTransactions().get(0));
        stateManager.update(firstAppend);
        stateManager.onHighWatermarkUpdated(21L);
        assertEquals(Optional.of(new LogOffsetMetadata(16L)), stateManager.firstUnstableOffset());

        // Now do a single append which completes the old transaction, mixes in
        // some empty transactions, one non-empty complete transaction, and one
        // incomplete transaction
        ProducerAppendInfo secondAppend = stateManager.prepareUpdate(producerId, AppendOrigin.CLIENT);
        Optional<CompletedTxn> firstCompletedTxn = appendEndTransaction(ControlRecordType.COMMIT, 21, secondAppend);
        assertTrue(firstCompletedTxn.isPresent());
        assertEquals(Optional.of(new CompletedTxn(producerId, 16L, 21, false)), firstCompletedTxn);
        assertFalse(appendEndTransaction(ControlRecordType.COMMIT, 22, secondAppend).isPresent());
        assertFalse(appendEndTransaction(ControlRecordType.ABORT, 23, secondAppend).isPresent());
        appendData(24L, 27L, secondAppend, seq);
        Optional<CompletedTxn> secondCompletedTxn = appendEndTransaction(ControlRecordType.ABORT, 28L, secondAppend);
        assertTrue(secondCompletedTxn.isPresent());
        assertFalse(appendEndTransaction(ControlRecordType.ABORT, 29L, secondAppend).isPresent());
        appendData(30L, 31L, secondAppend, seq);

        int size = secondAppend.startedTransactions().size();
        assertEquals(2, size);
        assertTxnMetadataEquals(new TxnMetadata(producerId, new LogOffsetMetadata(24L)), secondAppend.startedTransactions().get(0));
        assertTxnMetadataEquals(new TxnMetadata(producerId, new LogOffsetMetadata(30L)), secondAppend.startedTransactions().get(size - 1));
        stateManager.update(secondAppend);
        stateManager.completeTxn(firstCompletedTxn.get());
        stateManager.completeTxn(secondCompletedTxn.get());
        stateManager.onHighWatermarkUpdated(32L);
        assertEquals(Optional.of(new LogOffsetMetadata(30L)), stateManager.firstUnstableOffset());
    }
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