public long loadBatch(Batch<ApiMessageAndVersion> batch, LeaderAndEpoch leaderAndEpoch) {
        long startNs = time.nanoseconds();
        int indexWithinBatch = 0;

        lastContainedLogTimeMs = batch.appendTimestamp();
        lastEpoch = batch.epoch();

        for (ApiMessageAndVersion record : batch.records()) {
            try {
                replay(record);
            } catch (Throwable e) {
                faultHandler.handleFault("Error loading metadata log record from offset " +
                    (batch.baseOffset() + indexWithinBatch), e);
            }

            // Emit the accumulated delta if a new transaction has been started and one of the following is true
            // 1) this is not the first record in this batch
            // 2) this is not the first batch since last emitting a delta
            if (transactionState == TransactionState.STARTED_TRANSACTION && (indexWithinBatch > 0 || numBatches > 0)) {
                MetadataProvenance provenance = new MetadataProvenance(lastOffset, lastEpoch, lastContainedLogTimeMs);
                LogDeltaManifest manifest = LogDeltaManifest.newBuilder()
                    .provenance(provenance)
                    .leaderAndEpoch(leaderAndEpoch)
                    .numBatches(numBatches) // This will be zero if we have not yet read a batch
                    .elapsedNs(totalBatchElapsedNs)
                    .numBytes(numBytes)     // This will be zero if we have not yet read a batch
                    .build();
                if (log.isDebugEnabled()) {
                    log.debug("handleCommit: Generated a metadata delta between {} and {} from {} batch(es) in {} us.",
                            image.offset(), manifest.provenance().lastContainedOffset(),
                            manifest.numBatches(), NANOSECONDS.toMicros(manifest.elapsedNs()));
                }
                applyDeltaAndUpdate(delta, manifest);
                transactionState = TransactionState.STARTED_TRANSACTION;
            }

            lastOffset = batch.baseOffset() + indexWithinBatch;
            indexWithinBatch++;
        }
        long elapsedNs = time.nanoseconds() - startNs;

        // Update state for the manifest. The actual byte count will only be included in the last delta emitted for
        // a given batch or transaction.
        lastOffset = batch.lastOffset();
        numBytes += batch.sizeInBytes();
        numBatches += 1;
        totalBatchElapsedNs += elapsedNs;
        return totalBatchElapsedNs;
    }
Load a batch of records from the log. We have to do some bookkeeping here to
translate between batch offsets and record offsets, and track the number of bytes we
have read. Additionally, there is the chance that one of the records is a metadata
version change which needs to be handled differently.
</p>
If this batch starts a transaction, any records preceding the transaction in this
batch will be implicitly added to the transaction.

@param batch    The reader which yields the batches.
@return         The time in nanoseconds that elapsed while loading this batch
metadata/src/main/java/org/apache/kafka/image/loader/MetadataBatchLoader.java