Repository: apache/kafka
File: storage/src/test/java/org/apache/kafka/storage/internals/log/ProducerStateManagerTest.java (Java)

Code before:
        assertThrows(OutOfOrderSequenceException.class,
                () -> appendClientEntry(stateManager, producerId, epoch, 4, 0, true));

        assertEquals(originalEntry, stateManager.verificationStateEntry(producerId));
    }

    @Test
    public void testVerificationStateEntryExpiration() {
        VerificationStateEntry originalEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, epoch, true);

        // Before timeout, we do not remove. Note: Accessing the verification entry does not update the time.
        time.sleep(producerStateManagerConfig.producerIdExpirationMs() / 2);
        stateManager.removeExpiredProducers(time.milliseconds());
        assertEquals(originalEntry, stateManager.verificationStateEntry(producerId));

        time.sleep((producerStateManagerConfig.producerIdExpirationMs() / 2) + 1);
        stateManager.removeExpiredProducers(time.milliseconds());
        assertNull(stateManager.verificationStateEntry(producerId));
    }

    @Test
    public void testIdempotentTransactionMarkerExceptionThrownTV2() {
        short transactionVersion = 2;
        appendClientEntry(stateManager, producerId, epoch, defaultSequence, 99, true);
        assertEquals(OptionalLong.of(99L), stateManager.firstUndecidedOffset());

        short markerEpoch = (short) (epoch + 1);
        appendEndTxnMarker(stateManager, producerId, markerEpoch, ControlRecordType.COMMIT, 100, transactionVersion);

        ProducerStateEntry entry = getLastEntryOrElseThrownByProducerId(stateManager, producerId);
        assertEquals(markerEpoch, entry.producerEpoch());
        assertEquals(OptionalLong.empty(), entry.currentTxnFirstOffset());

        ProducerAppendInfo appendInfo = stateManager.prepareUpdate(producerId, AppendOrigin.COORDINATOR);
        EndTransactionMarker endTxnMarker = new EndTransactionMarker(ControlRecordType.COMMIT, 0);

        assertDoesNotThrow(
                () -> appendInfo.appendEndTxnMarker(endTxnMarker, markerEpoch, 101, time.milliseconds(), transactionVersion)
        );
    }

    @Test
    public void testRejectNonZeroSequenceForTransactionsV2WithEmptyState() {
        // Create a verification state entry that supports epoch bump (transactions v2)
        VerificationStateEntry verificationEntry = stateManager.maybeCreateVerificationStateEntry(
            producerId,
            0,
            epoch,
            true
        );
        

Review comment (refactor):
small question -- why do we do this append in two different ways in the test? It seems like the behavior is mostly the same in this helper vs the code at 1106. It's just the code at 1106 doesn't do three of the steps at the end:
```
private void appendEndTxnMarker(ProducerStateManager stateManager,
                                    long producerId,
                                    short producerEpoch,
                                    ControlRecordType controlType,
                                    long offset,
                                    int coordinatorEpoch,
                                    long timestamp,
                                    short transactionVersion) {
        ProducerAppendInfo producerAppendInfo = stateManager.prepareUpdate(producerId, AppendOrigin.COORDINATOR, time.milliseconds());
        EndTransactionMarker endTxnMarker = new EndTransactionMarker(controlType, coordinatorEpoch);
        Optional<CompletedTxn> completedTxn = producerAppendInfo.appendEndTxnMarker(endTxnMarker, producerEpoch, offset, timestamp, transactionVersion);
        stateManager.update(producerAppendInfo);
        completedTxn.ifPresent(stateManager::completeTxn);
        stateManager.updateMapEndOffset(offset + 1);
    }
```

Code after:
        assertThrows(OutOfOrderSequenceException.class,
                () -> appendClientEntry(stateManager, producerId, epoch, 4, 0, true));

        assertEquals(originalEntry, stateManager.verificationStateEntry(producerId));
    }

    @Test
    public void testVerificationStateEntryExpiration() {
        VerificationStateEntry originalEntry = stateManager.maybeCreateVerificationStateEntry(producerId, 0, epoch, true);

        // Before timeout, we do not remove. Note: Accessing the verification entry does not update the time.
        time.sleep(producerStateManagerConfig.producerIdExpirationMs() / 2);
        stateManager.removeExpiredProducers(time.milliseconds());
        assertEquals(originalEntry, stateManager.verificationStateEntry(producerId));

        time.sleep((producerStateManagerConfig.producerIdExpirationMs() / 2) + 1);
        stateManager.removeExpiredProducers(time.milliseconds());
        assertNull(stateManager.verificationStateEntry(producerId));
    }

    @Test
    public void testIdempotentTransactionMarkerRetryTV2() {
        short transactionVersion = 2;
        appendClientEntry(stateManager, producerId, epoch, defaultSequence, 99, true);
        short markerEpoch = (short) (epoch + 1);
        appendEndTxnMarker(stateManager, producerId, markerEpoch, ControlRecordType.COMMIT, 100, transactionVersion);

        ProducerStateEntry entry = getLastEntryOrElseThrownByProducerId(stateManager, producerId);
        assertEquals(markerEpoch, entry.producerEpoch());
        assertEquals(OptionalLong.empty(), entry.currentTxnFirstOffset());

        assertDoesNotThrow(() ->
                appendEndTxnMarker(stateManager, producerId, markerEpoch, ControlRecordType.COMMIT, 101, transactionVersion)
        );

        ProducerStateEntry entryAfterRetry = getLastEntryOrElseThrownByProducerId(stateManager, producerId);
        assertEquals(markerEpoch, entryAfterRetry.producerEpoch());
        assertEquals(OptionalLong.empty(), entryAfterRetry.currentTxnFirstOffset());
    }

    @Test
    public void testRejectNonZeroSequenceForTransactionsV2WithEmptyState() {
        // Create a verification state entry that supports epoch bump (transactions v2)
        VerificationStateEntry verificationEntry = stateManager.maybeCreateVerificationStateEntry(
            producerId,
            0,
            epoch,
            true
        );
        
        // Verify this is actually transactions v2