public void maybeFlushBatches(LeaderAndEpoch leaderAndEpoch) {
        MetadataProvenance provenance = new MetadataProvenance(lastOffset, lastEpoch, lastContainedLogTimeMs);
        LogDeltaManifest manifest = LogDeltaManifest.newBuilder()
            .provenance(provenance)
            .leaderAndEpoch(leaderAndEpoch)
            .numBatches(numBatches)
            .elapsedNs(totalBatchElapsedNs)
            .numBytes(numBytes)
            .build();
        switch (transactionState) {
            case STARTED_TRANSACTION:
            case CONTINUED_TRANSACTION:
                log.debug("handleCommit: not publishing since a transaction starting at {} is still in progress. " +
                    "{} batch(es) processed so far.", image.offset(), numBatches);
                break;
            case ABORTED_TRANSACTION:
                log.debug("handleCommit: publishing empty delta between {} and {} from {} batch(es) " +
                    "since a transaction was aborted", image.offset(), manifest.provenance().lastContainedOffset(),
                    manifest.numBatches());
                applyDeltaAndUpdate(new MetadataDelta.Builder().setImage(image).build(), manifest);
                break;
            case ENDED_TRANSACTION:
            case NO_TRANSACTION:
                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);
                break;
        }
    }
Flush the metadata accumulated in this batch loader if not in the middle of a transaction. The
flushed metadata will be passed to the {@link MetadataUpdater} configured for this class.
metadata/src/main/java/org/apache/kafka/image/loader/MetadataBatchLoader.java