MINOR: Refactor the MetadataPublisher interface

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

@@ -26,7 +26,7 @@ import kafka.network.{DataPlaneAcceptor, SocketServer}
 import kafka.raft.KafkaRaftManager
 import kafka.security.CredentialProvider
 import kafka.server.KafkaRaftServer.ControllerRole
-import kafka.server.metadata.{BrokerMetadataListener, BrokerMetadataPublisher, BrokerMetadataSnapshotter, ClientQuotaMetadataManager, DynamicConfigPublisher, KRaftMetadataCache, SnapshotWriterBuilder}
+import kafka.server.metadata.{BrokerMetadataListener, BrokerMetadataPublisher, BrokerMetadataSnapshotter, ClientQuotaMetadataManager, DynamicClientQuotaPublisher, DynamicConfigPublisher, KRaftMetadataCache, ScramPublisher, SnapshotWriterBuilder}
 import kafka.utils.CoreUtils
 import org.apache.kafka.common.feature.SupportedVersionRange
 import org.apache.kafka.common.message.ApiMessageType.ListenerType
@@ -217,8 +217,9 @@ class BrokerServer(
       tokenCache = new DelegationTokenCache(ScramMechanism.mechanismNames)
       credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, tokenCache)
 
-      val voterConnections = FutureUtils.waitWithLogging(logger.underlying,
-        "controller quorum voters future", sharedServer.controllerQuorumVotersFuture,
+      val voterConnections = FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "controller quorum voters future",
+        sharedServer.controllerQuorumVotersFuture,
         startupDeadline, time)
       val controllerNodes = RaftConfig.voterConnectionsToNodes(voterConnections).asScala
       val controllerNodeProvider = RaftControllerNodeProvider(raftManager, config, controllerNodes)
@@ -437,7 +438,8 @@ class BrokerServer(
         config.numIoThreads, s"${DataPlaneAcceptor.MetricPrefix}RequestHandlerAvgIdlePercent",
         DataPlaneAcceptor.ThreadPrefix)
 
-      FutureUtils.waitWithLogging(logger.underlying, "broker metadata to catch up",
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "broker metadata to catch up",
         lifecycleManager.initialCatchUpFuture, startupDeadline, time)
 
       // Apply the metadata log changes that we've accumulated.
@@ -447,14 +449,22 @@ class BrokerServer(
         replicaManager,
         groupCoordinator,
         transactionCoordinator,
-        clientQuotaMetadataManager,
         new DynamicConfigPublisher(
           config,
           sharedServer.metadataPublishingFaultHandler,
           dynamicConfigHandlers.toMap,
         "broker"),
+        new DynamicClientQuotaPublisher(
+          config,
+          sharedServer.metadataPublishingFaultHandler,
+          "broker",
+          clientQuotaMetadataManager),
+        new ScramPublisher(
+          config,
+          sharedServer.metadataPublishingFaultHandler,
+          "broker",
+          credentialProvider),
         authorizer,
-        credentialProvider,
         sharedServer.initialBrokerMetadataLoadFaultHandler,
         sharedServer.metadataPublishingFaultHandler)
 
@@ -465,7 +475,7 @@ class BrokerServer(
       // publish operation to complete. This first operation will initialize logManager,
       // replicaManager, groupCoordinator, and txnCoordinator. The log manager may perform
       // a potentially lengthy recovery-from-unclean-shutdown operation here, if required.
-      FutureUtils.waitWithLogging(logger.underlying,
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
         "the broker to catch up with the current cluster metadata",
         metadataListener.startPublishing(metadataPublisher), startupDeadline, time)
 
@@ -489,15 +499,18 @@ class BrokerServer(
 
       // We're now ready to unfence the broker. This also allows this broker to transition
       // from RECOVERY state to RUNNING state, once the controller unfences the broker.
-      FutureUtils.waitWithLogging(logger.underlying, "the broker to be unfenced",
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "the broker to be unfenced",
         lifecycleManager.setReadyToUnfence(), startupDeadline, time)
 
       // Block here until all the authorizer futures are complete
-      FutureUtils.waitWithLogging(logger.underlying, "all of the authorizer futures to be completed",
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "all of the authorizer futures to be completed",
         CompletableFuture.allOf(authorizerFutures.values.toSeq: _*), startupDeadline, time)
 
       // Wait for all the SocketServer ports to be open, and the Acceptors to be started.
-      FutureUtils.waitWithLogging(logger.underlying, "all of the SocketServer Acceptors to be started",
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "all of the SocketServer Acceptors to be started",
         socketServerFuture, startupDeadline, time)
 
       maybeChangeStatus(STARTING, STARTED)

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

@@ -27,7 +27,7 @@ import kafka.server.KafkaConfig.{AlterConfigPolicyClassNameProp, CreateTopicPoli
 import kafka.server.QuotaFactory.QuotaManagers
 
 import scala.collection.immutable
-import kafka.server.metadata.{ClientQuotaMetadataManager, DynamicClientQuotaPublisher, DynamicConfigPublisher}
+import kafka.server.metadata.{ClientQuotaMetadataManager, DynamicClientQuotaPublisher, DynamicConfigPublisher, ScramPublisher}
 import kafka.utils.{CoreUtils, Logging}
 import kafka.zk.{KafkaZkClient, ZkMigrationClient}
 import org.apache.kafka.common.config.ConfigException
@@ -37,6 +37,7 @@ import org.apache.kafka.common.security.token.delegation.internals.DelegationTok
 import org.apache.kafka.common.utils.LogContext
 import org.apache.kafka.common.{ClusterResource, Endpoint}
 import org.apache.kafka.controller.{Controller, QuorumController, QuorumFeatures}
+import org.apache.kafka.image.publisher.MetadataPublisher
 import org.apache.kafka.metadata.KafkaConfigSchema
 import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer
 import org.apache.kafka.metadata.bootstrap.BootstrapMetadata
@@ -196,7 +197,7 @@ class ControllerServer(
       alterConfigPolicy = Option(config.
         getConfiguredInstance(AlterConfigPolicyClassNameProp, classOf[AlterConfigPolicy]))
 
-      val voterConnections = FutureUtils.waitWithLogging(logger.underlying,
+      val voterConnections = FutureUtils.waitWithLogging(logger.underlying, logIdent,
         "controller quorum voters future",
         sharedServer.controllerQuorumVotersFuture,
         startupDeadline, time)
@@ -297,33 +298,51 @@ class ControllerServer(
       val socketServerFuture = socketServer.enableRequestProcessing(authorizerFutures)
 
       // Block here until all the authorizer futures are complete
-      FutureUtils.waitWithLogging(logger.underlying, "all of the authorizer futures to be completed",
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "all of the authorizer futures to be completed",
         CompletableFuture.allOf(authorizerFutures.values.toSeq: _*), startupDeadline, time)
 
       // Wait for all the SocketServer ports to be open, and the Acceptors to be started.
-      FutureUtils.waitWithLogging(logger.underlying, "all of the SocketServer Acceptors to be started",
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "all of the SocketServer Acceptors to be started",
         socketServerFuture, startupDeadline, time)
 
       // register this instance for dynamic config changes to the KafkaConfig
       config.dynamicConfig.addReconfigurables(this)
-      // We must install the below publisher and receive the changes when we are also running the broker role
-      // because we don't share a single KafkaConfig instance with the broker, and therefore
-      // the broker's DynamicConfigPublisher won't take care of the changes for us.
-      val dynamicConfigHandlers = immutable.Map[String, ConfigHandler](
-        // controllers don't host topics, so no need to do anything with dynamic topic config changes here
-        ConfigType.Broker -> new BrokerConfigHandler(config, quotaManagers))
-      val dynamicConfigPublisher = new DynamicConfigPublisher(
+
+      val publishers = new java.util.ArrayList[MetadataPublisher]()
+
+      // Set up the dynamic config publisher. This runs even in combined mode, since the broker
+      // has its own separate dynamic configuration object.
+      publishers.add(new DynamicConfigPublisher(
         config,
         sharedServer.metadataPublishingFaultHandler,
-        dynamicConfigHandlers,
-        "controller")
-      val dynamicClientQuotaPublisher = new DynamicClientQuotaPublisher(
+        immutable.Map[String, ConfigHandler](
+          // controllers don't host topics, so no need to do anything with dynamic topic config changes here
+          ConfigType.Broker -> new BrokerConfigHandler(config, quotaManagers)
+        ),
+        "controller"))
+
+      // Set up the client quotas publisher. This will enable controller mutation quotas and any
+      // other quotas which are applicable.
+      publishers.add(new DynamicClientQuotaPublisher(
         config,
         sharedServer.metadataPublishingFaultHandler,
         "controller",
-        clientQuotaMetadataManager)
-      FutureUtils.waitWithLogging(logger.underlying, "all of the dynamic config and client quota publishers to be installed",
-        sharedServer.loader.installPublishers(List(dynamicConfigPublisher, dynamicClientQuotaPublisher).asJava), startupDeadline, time)
+        clientQuotaMetadataManager))
+
+      // Set up the SCRAM publisher.
+      publishers.add(new ScramPublisher(
+        config,
+        sharedServer.metadataPublishingFaultHandler,
+        "controller",
+        credentialProvider
+      ))
+
+      // Install all metadata publishers.
+      FutureUtils.waitWithLogging(logger.underlying, logIdent,
+        "all of the metadata publishers to be installed",
+        sharedServer.loader.installPublishers(publishers), startupDeadline, time)
     } catch {
       case e: Throwable =>
         maybeChangeStatus(STARTING, STARTED)

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

@@ -21,7 +21,6 @@ import java.util.{OptionalInt, Properties}
 import java.util.concurrent.atomic.AtomicLong
 import kafka.coordinator.transaction.TransactionCoordinator
 import kafka.log.{LogManager, UnifiedLog}
-import kafka.security.CredentialProvider
 import kafka.server.{KafkaConfig, ReplicaManager, RequestLocal}
 import kafka.utils.Logging
 import org.apache.kafka.common.TopicPartition
@@ -102,10 +101,10 @@ class BrokerMetadataPublisher(
   replicaManager: ReplicaManager,
   groupCoordinator: GroupCoordinator,
   txnCoordinator: TransactionCoordinator,
-  clientQuotaMetadataManager: ClientQuotaMetadataManager,
   var dynamicConfigPublisher: DynamicConfigPublisher,
+  dynamicClientQuotaPublisher: DynamicClientQuotaPublisher,
+  scramPublisher: ScramPublisher,
   private val _authorizer: Option[Authorizer],
-  credentialProvider: CredentialProvider,
   fatalFaultHandler: FaultHandler,
   metadataPublishingFaultHandler: FaultHandler
 ) extends MetadataPublisher with Logging {
@@ -213,32 +212,13 @@ class BrokerMetadataPublisher(
       }
 
       // Apply configuration deltas.
-      dynamicConfigPublisher.publish(delta, newImage)
+      dynamicConfigPublisher.onMetadataUpdate(delta, newImage)
 
       // Apply client quotas delta.
-      try {
-        Option(delta.clientQuotasDelta()).foreach { clientQuotasDelta =>
-          clientQuotaMetadataManager.update(clientQuotasDelta)
-        }
-      } catch {
-        case t: Throwable => metadataPublishingFaultHandler.handleFault("Error updating client " +
-          s"quotas in $deltaName", t)
-      }
+      dynamicClientQuotaPublisher.onMetadataUpdate(delta, newImage)
 
-      // Apply changes to SCRAM credentials.
-      Option(delta.scramDelta()).foreach { scramDelta =>
-        scramDelta.changes().forEach {
-          case (mechanism, userChanges) =>
-            userChanges.forEach {
-              case (userName, change) =>
-                if (change.isPresent) {
-                  credentialProvider.updateCredential(mechanism, userName, change.get().toCredential(mechanism))
-                } else {
-                  credentialProvider.removeCredentials(mechanism, userName)
-                }
-            }
-        }
-      }
+      // Apply SCRAM delta.
+      scramPublisher.onMetadataUpdate(delta, newImage)
 
       // Apply changes to ACLs. This needs to be handled carefully because while we are
       // applying these changes, the Authorizer is continuing to return authorization

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,20 @@ 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}"
+
+  override def onMetadataUpdate(
+    delta: MetadataDelta,
+    newImage: MetadataImage,
+    manifest: LoaderManifest
+  ): Unit = {
+    onMetadataUpdate(delta, newImage)
+  }
+
+  def onMetadataUpdate(
+    delta: MetadataDelta,
+    newImage: MetadataImage,
+  ): Unit = {
     val deltaName = s"MetadataDelta up to ${newImage.highestOffsetAndEpoch().offset}"
     try {
         Option(delta.clientQuotasDelta()).foreach { clientQuotasDelta =>
@@ -43,40 +56,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,20 @@ 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}"
+
+  override def onMetadataUpdate(
+    delta: MetadataDelta,
+    newImage: MetadataImage,
+    manifest: LoaderManifest
+  ): Unit = {
+    onMetadataUpdate(delta, newImage)
+  }
+
+  def onMetadataUpdate(
+    delta: MetadataDelta,
+    newImage: MetadataImage,
+  ): Unit = {
     val deltaName = s"MetadataDelta up to ${newImage.highestOffsetAndEpoch().offset}"
     try {
       // Apply configuration deltas.
@@ -101,40 +114,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
-  }
 }