public ProducerAppendInfo prepareUpdate(long producerId, AppendOrigin origin) {
        ProducerStateEntry currentEntry = lastEntry(producerId).orElse(ProducerStateEntry.empty(producerId));
        return new ProducerAppendInfo(topicPartition, producerId, currentEntry, origin, verificationStateEntry(producerId));
    }
@Test
    public void updateProducerTransactionState() {
        int coordinatorEpoch = 15;
        long offset = 9L;
        appendClientEntry(stateManager, producerId, epoch, defaultSequence, offset, false);

        ProducerAppendInfo appendInfo = stateManager.prepareUpdate(producerId, AppendOrigin.CLIENT);
        appendInfo.appendDataBatch(epoch, 1, 5, time.milliseconds(),
                new LogOffsetMetadata(16L), 20L, true);
        verifyLastEntryWithTxnData(appendInfo.toEntry(), 1, 5,
                16L, 20L, OptionalLong.of(16), appendInfo);

        appendInfo.appendDataBatch(epoch, 6, 10, time.milliseconds(),
                new LogOffsetMetadata(26L), 30L, true);
        verifyLastEntryWithTxnData(appendInfo.toEntry(), 1, 10,
                16L, 30L, OptionalLong.of(16), appendInfo);

        EndTransactionMarker endTxnMarker = new EndTransactionMarker(ControlRecordType.COMMIT, coordinatorEpoch);
        CompletedTxn completedTxn = appendInfo.appendEndTxnMarker(endTxnMarker, epoch, 40L, time.milliseconds())
                .orElseThrow(() -> new RuntimeException("The transaction should be completed"));

        assertEquals(producerId, completedTxn.producerId);
        assertEquals(16L, completedTxn.firstOffset);
        assertEquals(40L, completedTxn.lastOffset);
        assertFalse(completedTxn.isAborted);

        ProducerStateEntry lastEntry = appendInfo.toEntry();
        // verify that appending the transaction marker doesn't affect the metadata of the cached record batches.
        verifyLastEntryWithTxnData(lastEntry, 1, 10,
                16L, 30L, OptionalLong.empty(), appendInfo);
        assertEquals(OptionalLong.empty(), lastEntry.currentTxnFirstOffset());
    }