private void updateProducerState(ProducerStateManager producerStateManager, RecordBatch batch) throws IOException {
        if (batch.hasProducerId()) {
            long producerId = batch.producerId();
            ProducerAppendInfo appendInfo = producerStateManager.prepareUpdate(producerId, AppendOrigin.REPLICATION);
            Optional<CompletedTxn> maybeCompletedTxn = appendInfo.append(batch, Optional.empty());
            producerStateManager.update(appendInfo);
            if (maybeCompletedTxn.isPresent()) {
                CompletedTxn completedTxn = maybeCompletedTxn.get();
                long lastStableOffset = producerStateManager.lastStableOffset(completedTxn);
                updateTxnIndex(completedTxn, lastStableOffset);
                producerStateManager.completeTxn(completedTxn);
            }
        }
        producerStateManager.updateMapEndOffset(batch.lastOffset() + 1);
    }
Append records from a file beginning at the given position until either the end of the file
is reached or an offset is found which is too large to convert to a relative offset for the indexes.

@return the number of bytes appended to the log (may be less than the size of the input if an
        offset is encountered which would overflow this segment)
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java