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;
        }
    }
@Test
    public void testMultipleTransactionsInOneBatchesWithNoOp() {
        List<ApiMessageAndVersion> batchRecords = new ArrayList<>();
        batchRecords.addAll(noOpRecords(1));
        batchRecords.addAll(TOPIC_TXN_BATCH_1);
        batchRecords.addAll(noOpRecords(1));
        batchRecords.addAll(TOPIC_TXN_BATCH_2);
        // A batch with non-transactional records between two transactions causes a delta to get published
        batchRecords.addAll(noOpRecords(1));
        batchRecords.addAll(TXN_BEGIN_SINGLETON);
        batchRecords.addAll(noOpRecords(1));
        batchRecords.addAll(TOPIC_NO_TXN_BATCH);
        batchRecords.addAll(noOpRecords(1));
        batchRecords.addAll(TXN_END_SINGLETON);
        batchRecords.addAll(noOpRecords(1));

        MockMetadataUpdater updater = new MockMetadataUpdater();
        MockFaultHandler faultHandler = new MockFaultHandler("testMultipleTransactionsInOneBatches");
        MetadataBatchLoader batchLoader = loadSingleBatch(updater, faultHandler, batchRecords);

        assertEquals(2, updater.updates);
        assertEquals(0, updater.latestManifest.numBytes());
        assertEquals(18, updater.latestImage.provenance().lastContainedOffset());
        assertEquals(42, updater.latestImage.provenance().lastContainedEpoch());
        assertNotNull(updater.latestImage.topics().getTopic("foo"));
        assertNull(updater.latestImage.topics().getTopic("bar"));
        batchLoader.maybeFlushBatches(LEADER_AND_EPOCH);
        assertEquals(3, updater.updates);
        assertEquals(100, updater.latestManifest.numBytes());
        assertEquals(26, updater.latestImage.provenance().lastContainedOffset());
        assertEquals(42, updater.latestImage.provenance().lastContainedEpoch());
        assertNotNull(updater.latestImage.topics().getTopic("foo"));
        assertNotNull(updater.latestImage.topics().getTopic("bar"));
    }