@Test
    public void testLastStableOffsetCompletedTxn() {
        long segmentBaseOffset = 990000L;

        long producerId1 = producerId;
        long startOffset1 = 992342L;
        beginTransaction(producerId1, startOffset1, segmentBaseOffset);

        long producerId2 = producerId + 1;
        long startOffset2 = startOffset1 + 25;
        beginTransaction(producerId2, startOffset2, segmentBaseOffset);

        long producerId3 = producerId + 2;
        long startOffset3 = startOffset1 + 57;
        beginTransaction(producerId3, startOffset3, segmentBaseOffset);

        long lastOffset1 = startOffset3 + 15;
        CompletedTxn completedTxn1 = new CompletedTxn(producerId1, startOffset1, lastOffset1, false);
        assertEquals(startOffset2, stateManager.lastStableOffset(completedTxn1));
        stateManager.completeTxn(completedTxn1);
        stateManager.onHighWatermarkUpdated(lastOffset1 + 1);

        assertEquals(Optional.of(startOffset2), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        long lastOffset3 = lastOffset1 + 20;
        CompletedTxn completedTxn3 = new CompletedTxn(producerId3, startOffset3, lastOffset3, false);
        assertEquals(startOffset2, stateManager.lastStableOffset(completedTxn3));
        stateManager.completeTxn(completedTxn3);
        stateManager.onHighWatermarkUpdated(lastOffset3 + 1);
        assertEquals(Optional.of(startOffset2), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        long lastOffset2 = lastOffset3 + 78;
        CompletedTxn completedTxn2 = new CompletedTxn(producerId2, startOffset2, lastOffset2, false);
        assertEquals(lastOffset2 + 1, stateManager.lastStableOffset(completedTxn2));
        stateManager.completeTxn(completedTxn2);
        stateManager.onHighWatermarkUpdated(lastOffset2 + 1);
        assertEquals(Optional.empty(), 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