private void maybePublishMetadata(MetadataDelta delta, MetadataImage image, LoaderManifest manifest) {
        // AutoMQ inject start
        MetadataImage oldImage = this.image;
        // AutoMQ inject end
        this.image = image;

        if (stillNeedToCatchUp(
            "maybePublishMetadata(" + manifest.type().toString() + ")",
            manifest.provenance().lastContainedOffset())
        ) {
            // AutoMQ inject start
            tryReleaseOldImage(oldImage, image);
            // AutoMQ inject end
            return;
        }

        if (log.isTraceEnabled()) {
            log.trace("handleCommit: publishing new image with provenance {}.", image.provenance());
        }
        for (MetadataPublisher publisher : publishers.values()) {
            try {
                publisher.onMetadataUpdate(delta, image, manifest);
            } catch (Throwable e) {
                faultHandler.handleFault("Unhandled error publishing the new metadata " +
                    "image ending at " + manifest.provenance().lastContainedOffset() +
                    " with publisher " + publisher.name(), e);
            }
        }
        metrics.updateLastAppliedImageProvenance(image.provenance());
        metrics.setCurrentMetadataVersion(image.features().metadataVersion());
        if (!uninitializedPublishers.isEmpty()) {
            scheduleInitializeNewPublishers(0);
        }
        // AutoMQ inject start
        tryReleaseOldImage(oldImage, image);
        // AutoMQ inject end
    }
Callback used by MetadataBatchLoader and handleLoadSnapshot to update the active metadata publishers.
metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java