@Override
    public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
        eventQueue.append(() -> {
            try {
                while (reader.hasNext()) {
                    Batch<ApiMessageAndVersion> batch = reader.next();
                    long elapsedNs = batchLoader.loadBatch(batch, currentLeaderAndEpoch);
                    metrics.updateBatchSize(batch.records().size());
                    metrics.updateBatchProcessingTimeNs(elapsedNs);
                }
                batchLoader.maybeFlushBatches(currentLeaderAndEpoch);
            } catch (Throwable e) {
                // This is a general catch-all block where we don't expect to end up;
                // failure-prone operations should have individual try/catch blocks around them.
                faultHandler.handleFault("Unhandled fault in MetadataLoader#handleCommit. " +
                    "Last image offset was " + image.offset(), e);
            } finally {
                reader.close();
            }
        });
    }
Callback used by MetadataBatchLoader and handleLoadSnapshot to update the active metadata publishers.
metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java