void initializeNewPublishers() {
        if (uninitializedPublishers.isEmpty()) {
            log.debug("InitializeNewPublishers: nothing to do.");
            return;
        }
        if (stillNeedToCatchUp("initializeNewPublishers", image.highestOffsetAndEpoch().offset())) {
            // Reschedule the initialization for later.
            log.debug("InitializeNewPublishers: unable to initialize new publisher(s) {} " +
                            "because we are still catching up with quorum metadata. Rescheduling.",
                    uninitializedPublisherNames());
            scheduleInitializeNewPublishers(TimeUnit.MILLISECONDS.toNanos(100));
            return;
        }
        log.debug("InitializeNewPublishers: setting up snapshot image for new publisher(s): {}",
                uninitializedPublisherNames());
        long startNs = time.nanoseconds();
        // We base this delta off of the empty image, reflecting the fact that these publishers
        // haven't seen anything previously.
        MetadataDelta delta = new MetadataDelta.Builder().
                setImage(MetadataImage.EMPTY).
                build();
        ImageReWriter writer = new ImageReWriter(delta);
        image.write(writer, new ImageWriterOptions.Builder().
                setMetadataVersion(image.features().metadataVersion()).
                build());
        // ImageReWriter#close invokes finishSnapshot, so we don't need to invoke it here.
        SnapshotManifest manifest = new SnapshotManifest(
                image.provenance(),
                time.nanoseconds() - startNs);
        for (Iterator<MetadataPublisher> iter = uninitializedPublishers.values().iterator();
                iter.hasNext(); ) {
            MetadataPublisher publisher = iter.next();
            iter.remove();
            try {
                log.info("InitializeNewPublishers: initializing {} with a snapshot at offset {}",
                        publisher.name(), image.highestOffsetAndEpoch().offset());
                publisher.onMetadataUpdate(delta, image, manifest);
                publisher.onControllerChange(currentLeaderAndEpoch);
                publishers.put(publisher.name(), publisher);
            } catch (Throwable e) {
                faultHandler.handleFault("Unhandled error initializing " + publisher.name() +
                        " with a snapshot at offset " + image.highestOffsetAndEpoch().offset(), e);
            }
        }
    }
Schedule an event to initialize the new publishers that are present in the system.

@param delayNs   The minimum time in nanoseconds we should wait. If there is already an
                 initialization event scheduled, we will either move its deadline forward
                 in time or leave it unchanged.
metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java