MINOR: Refactor the MetadataPublisher interface

core/src/main/scala/kafka/server/KafkaRaftServer.scala

@@ -77,10 +77,7 @@ class KafkaRaftServer(
   )
 
   private val broker: Option[BrokerServer] = if (config.processRoles.contains(BrokerRole)) {
-    Some(new BrokerServer(
-      sharedServer,
-      offlineDirs
-    ))
+    Some(new BrokerServer(sharedServer, offlineDirs))
   } else {
     None
   }

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala

@@ -213,7 +213,7 @@ class BrokerMetadataPublisher(
       }
 
       // Apply configuration deltas.
-      dynamicConfigPublisher.publish(delta, newImage)
+      dynamicConfigPublisher.publish(delta, newImage, null)
 
       // Apply client quotas delta.
       try {

core/src/main/scala/kafka/server/metadata/DynamicClientQuotaPublisher.scala

@@ -19,7 +19,7 @@ package kafka.server.metadata
 
 import kafka.server.KafkaConfig
 import kafka.utils.Logging
-import org.apache.kafka.image.loader.{LogDeltaManifest, SnapshotManifest}
+import org.apache.kafka.image.loader.LoaderManifest
 import org.apache.kafka.image.{MetadataDelta, MetadataImage}
 import org.apache.kafka.server.fault.FaultHandler
 
@@ -32,7 +32,13 @@ class DynamicClientQuotaPublisher(
 ) extends Logging with org.apache.kafka.image.publisher.MetadataPublisher {
   logIdent = s"[${name()}] "
 
-  def publish(delta: MetadataDelta, newImage: MetadataImage): Unit = {
+  override def name(): String = s"DynamicClientQuotaPublisher ${nodeType} id=${conf.nodeId}"
+
+  def publish(
+    delta: MetadataDelta,
+    newImage: MetadataImage,
+    manifest: LoaderManifest
+  ): Unit = {
     val deltaName = s"MetadataDelta up to ${newImage.highestOffsetAndEpoch().offset}"
     try {
         Option(delta.clientQuotasDelta()).foreach { clientQuotasDelta =>
@@ -43,40 +49,4 @@ class DynamicClientQuotaPublisher(
         s"publishing dynamic client quota changes from ${deltaName}", t)
     }
   }
-
-  /**
-   * Returns the name of this publisher.
-   *
-   * @return The publisher name.
-   */
-  override def name(): String = s"DynamicClientQuotaPublisher ${nodeType} id=${conf.nodeId}"
-
-  /**
-   * Publish a new cluster metadata snapshot that we loaded.
-   *
-   * @param delta    The delta between the previous state and the new one.
-   * @param newImage The complete new state.
-   * @param manifest The contents of what was published.
-   */
-  override def publishSnapshot(delta: MetadataDelta, newImage: MetadataImage, manifest: SnapshotManifest): Unit = {
-    publish(delta, newImage)
-  }
-
-  /**
-   * Publish a change to the cluster metadata.
-   *
-   * @param delta    The delta between the previous state and the new one.
-   * @param newImage The complete new state.
-   * @param manifest The contents of what was published.
-   */
-  override def publishLogDelta(delta: MetadataDelta, newImage: MetadataImage, manifest: LogDeltaManifest): Unit = {
-    publish(delta, newImage)
-  }
-
-  /**
-   * Close this metadata publisher.
-   */
-  override def close(): Unit = {
-    // nothing to close
-  }
 }

core/src/main/scala/kafka/server/metadata/DynamicConfigPublisher.scala

@@ -22,7 +22,7 @@ import kafka.server.ConfigAdminManager.toLoggableProps
 import kafka.server.{ConfigEntityName, ConfigHandler, ConfigType, KafkaConfig}
 import kafka.utils.Logging
 import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, TOPIC}
-import org.apache.kafka.image.loader.{LogDeltaManifest, SnapshotManifest}
+import org.apache.kafka.image.loader.LoaderManifest
 import org.apache.kafka.image.{MetadataDelta, MetadataImage}
 import org.apache.kafka.server.fault.FaultHandler
 
@@ -35,7 +35,13 @@ class DynamicConfigPublisher(
 ) extends Logging with org.apache.kafka.image.publisher.MetadataPublisher {
   logIdent = s"[${name()}] "
 
-  def publish(delta: MetadataDelta, newImage: MetadataImage): Unit = {
+  override def name(): String = s"DynamicConfigPublisher ${nodeType} id=${conf.nodeId}"
+
+  def publish(
+    delta: MetadataDelta,
+    newImage: MetadataImage,
+    manifest: LoaderManifest
+  ): Unit = {
     val deltaName = s"MetadataDelta up to ${newImage.highestOffsetAndEpoch().offset}"
     try {
       // Apply configuration deltas.
@@ -101,40 +107,4 @@ class DynamicConfigPublisher(
   def reloadUpdatedFilesWithoutConfigChange(props: Properties): Unit = {
     conf.dynamicConfig.reloadUpdatedFilesWithoutConfigChange(props)
   }
-
-  /**
-   * Returns the name of this publisher.
-   *
-   * @return The publisher name.
-   */
-  override def name(): String = s"DynamicConfigPublisher ${nodeType} id=${conf.nodeId}"
-
-  /**
-   * Publish a new cluster metadata snapshot that we loaded.
-   *
-   * @param delta    The delta between the previous state and the new one.
-   * @param newImage The complete new state.
-   * @param manifest The contents of what was published.
-   */
-  override def publishSnapshot(delta: MetadataDelta, newImage: MetadataImage, manifest: SnapshotManifest): Unit = {
-    publish(delta, newImage)
-  }
-
-  /**
-   * Publish a change to the cluster metadata.
-   *
-   * @param delta    The delta between the previous state and the new one.
-   * @param newImage The complete new state.
-   * @param manifest The contents of what was published.
-   */
-  override def publishLogDelta(delta: MetadataDelta, newImage: MetadataImage, manifest: LogDeltaManifest): Unit = {
-    publish(delta, newImage)
-  }
-
-  /**
-   * Close this metadata publisher.
-   */
-  override def close(): Unit = {
-    // nothing to close
-  }
 }

metadata/src/main/java/org/apache/kafka/image/loader/LoaderManifest.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.image.loader;
+
+import org.apache.kafka.image.MetadataProvenance;
+
+
+/**
+ * Contains information about what was loaded.
+ */
+public interface LoaderManifest {
+    /**
+     * Describes the type of manifest which this is.
+     */
+    LoaderManifestType type();
+
+    /**
+     * The highest offset and epoch included in the new image, inclusive.
+     */
+    MetadataProvenance provenance();
+}

metadata/src/main/java/org/apache/kafka/image/loader/LoaderManifestType.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.image.loader;
+
+
+/**
+ * Contains information about the type of a loader manifest.
+ */
+public enum LoaderManifestType {
+    LOG_DELTA,
+    SNAPSHOT;
+}

metadata/src/main/java/org/apache/kafka/image/loader/LogDeltaManifest.java

@@ -26,7 +26,7 @@
 /**
  * Contains information about a set of changes that were loaded from the metadata log.
  */
-public class LogDeltaManifest {
+public class LogDeltaManifest implements LoaderManifest {
     /**
      * The highest offset and epoch included in this delta, inclusive.
      */
@@ -66,7 +66,12 @@ public LogDeltaManifest(
         this.numBytes = numBytes;
     }
 
+    @Override
+    public LoaderManifestType type() {
+        return LoaderManifestType.LOG_DELTA;
+    }
 
+    @Override
     public MetadataProvenance provenance() {
         return provenance;
     }

metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java

@@ -258,7 +258,8 @@ private void maybeInitializeNewPublishers() {
             try {
                 log.info("Publishing initial snapshot at offset {} to {}",
                         image.highestOffsetAndEpoch().offset(), publisher.name());
-                publisher.publishSnapshot(delta, image, manifest);
+                publisher.publish(delta, image, manifest);
+                publisher.handleControllerChange(currentLeaderAndEpoch);
                 publishers.put(publisher.name(), publisher);
             } catch (Throwable e) {
                 faultHandler.handleFault("Unhandled error publishing the initial metadata " +
@@ -295,7 +296,7 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
                 log.debug("Publishing new image with provenance {}.", image.provenance());
                 for (MetadataPublisher publisher : publishers.values()) {
                     try {
-                        publisher.publishLogDelta(delta, image, manifest);
+                        publisher.publish(delta, image, manifest);
                     } catch (Throwable e) {
                         faultHandler.handleFault("Unhandled error publishing the new metadata " +
                             "image ending at " + manifest.provenance().lastContainedOffset() +
@@ -392,7 +393,7 @@ public void handleSnapshot(SnapshotReader<ApiMessageAndVersion> reader) {
                 log.debug("Publishing new snapshot image with provenance {}.", image.provenance());
                 for (MetadataPublisher publisher : publishers.values()) {
                     try {
-                        publisher.publishSnapshot(delta, image, manifest);
+                        publisher.publish(delta, image, manifest);
                     } catch (Throwable e) {
                         faultHandler.handleFault("Unhandled error publishing the new metadata " +
                                 "image from snapshot at offset " + reader.lastContainedLogOffset() +
@@ -449,6 +450,15 @@ SnapshotManifest loadSnapshot(
     public void handleLeaderChange(LeaderAndEpoch leaderAndEpoch) {
         eventQueue.append(() -> {
             currentLeaderAndEpoch = leaderAndEpoch;
+            for (MetadataPublisher publisher : publishers.values()) {
+                try {
+                    publisher.handleControllerChange(currentLeaderAndEpoch);
+                } catch (Throwable e) {
+                    faultHandler.handleFault("Unhandled error publishing the new leader " +
+                            "change to " + currentLeaderAndEpoch + " with publisher " +
+                            publisher.name(), e);
+                }
+            }
         });
     }
 

metadata/src/main/java/org/apache/kafka/image/loader/SnapshotManifest.java

@@ -25,7 +25,7 @@
 /**
  * Contains information about a snapshot that was loaded.
  */
-public class SnapshotManifest {
+public class SnapshotManifest implements LoaderManifest {
     /**
      * The source of this snapshot.
      */
@@ -44,6 +44,12 @@ public SnapshotManifest(
         this.elapsedNs = elapsedNs;
     }
 
+    @Override
+    public LoaderManifestType type() {
+        return LoaderManifestType.SNAPSHOT;
+    }
+
+    @Override
     public MetadataProvenance provenance() {
         return provenance;
     }

metadata/src/main/java/org/apache/kafka/image/publisher/MetadataPublisher.java

@@ -19,8 +19,8 @@
 
 import org.apache.kafka.image.MetadataDelta;
 import org.apache.kafka.image.MetadataImage;
-import org.apache.kafka.image.loader.LogDeltaManifest;
-import org.apache.kafka.image.loader.SnapshotManifest;
+import org.apache.kafka.image.loader.LoaderManifest;
+import org.apache.kafka.raft.LeaderAndEpoch;
 
 
 /**
@@ -40,33 +40,30 @@ public interface MetadataPublisher extends AutoCloseable {
     String name();
 
     /**
-     * Publish a new cluster metadata snapshot that we loaded.
+     * Handle a change in the current controller.
      *
-     * @param delta    The delta between the previous state and the new one.
-     * @param newImage The complete new state.
-     * @param manifest The contents of what was published.
+     * @param newLeaderAndEpoch The new quorum leader and epoch. The new leader will be
+     *                          OptionalInt.empty if there is currently no active controller.
      */
-    void publishSnapshot(
-            MetadataDelta delta,
-            MetadataImage newImage,
-            SnapshotManifest manifest
-    );
+    default void handleControllerChange(LeaderAndEpoch newLeaderAndEpoch) { }
 
     /**
-     * Publish a change to the cluster metadata.
+     * Publish a new cluster metadata snapshot that we loaded.
      *
      * @param delta    The delta between the previous state and the new one.
      * @param newImage The complete new state.
-     * @param manifest The contents of what was published.
+     * @param manifest A manifest which describes the contents of what was published.
+     *                 If we loaded a snapshot, this will be a SnapshotManifest.
+     *                 If we loaded a log delta, this will be a LogDeltaManifest.
      */
-    void publishLogDelta(
+    void publish(
             MetadataDelta delta,
             MetadataImage newImage,
-            LogDeltaManifest manifest
+            LoaderManifest manifest
     );
 
     /**
-     * Close this metadata publisher.
+     * Close this metadata publisher and free any associated resources.
      */
-    void close() throws Exception;
+    default void close() throws Exception { }
 }