public void completeTxn(CompletedTxn completedTxn) {
        TxnMetadata txnMetadata = ongoingTxns.remove(completedTxn.firstOffset);
        if (txnMetadata == null)
            throw new IllegalArgumentException("Attempted to complete transaction " + completedTxn + " on partition "
                    + topicPartition + " which was not started");

        txnMetadata.lastOffset = OptionalLong.of(completedTxn.lastOffset);
        unreplicatedTxns.put(completedTxn.firstOffset, txnMetadata);
        updateOldestTxnTimestamp();
    }
Mark a transaction as completed. We will still await advancement of the high watermark before
advancing the first unstable offset.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java