KAFKA-14327: Unify KRaft snapshot generation between broker and controller

checkstyle/import-control-core.xml

@@ -49,6 +49,7 @@
     <allow pkg="kafka.tools"/>
     <allow pkg="org.apache.kafka.clients"/>
     <allow pkg="org.apache.kafka.controller"/>
+    <allow pkg="org.apache.kafka.image"/>
     <allow pkg="org.apache.kafka.raft"/>
     <allow pkg="org.apache.kafka.test"/>
     <allow pkg="org.apache.kafka.metadata" />

checkstyle/import-control.xml

@@ -225,6 +225,7 @@
     <allow pkg="org.apache.kafka.common.requests" />
     <allow pkg="org.apache.kafka.common.resource" />
     <allow pkg="org.apache.kafka.controller" />
+    <allow pkg="org.apache.kafka.image.writer" />
     <allow pkg="org.apache.kafka.metadata" />
     <allow pkg="org.apache.kafka.metadata.authorizer" />
     <allow pkg="org.apache.kafka.metalog" />
@@ -252,10 +253,13 @@
     <allow pkg="org.apache.kafka.image" />
     <allow pkg="org.apache.kafka.image.writer" />
     <allow pkg="org.apache.kafka.metadata" />
+    <allow pkg="org.apache.kafka.queue" />
     <allow pkg="org.apache.kafka.raft" />
     <allow pkg="org.apache.kafka.server.common" />
+    <allow pkg="org.apache.kafka.server.fault" />
     <allow pkg="org.apache.kafka.server.util" />
     <allow pkg="org.apache.kafka.snapshot" />
+    <allow pkg="org.apache.kafka.test" />
   </subpackage>
 
   <subpackage name="metadata">
@@ -478,6 +482,10 @@
     <allow pkg="org.rocksdb" />
   </subpackage>
 
+  <subpackage name="queue">
+    <allow pkg="org.apache.kafka.test" />
+  </subpackage>
+
   <subpackage name="raft">
     <allow pkg="org.apache.kafka.raft" />
     <allow pkg="org.apache.kafka.metadata" />

checkstyle/suppressions.xml

@@ -38,9 +38,7 @@
     <!-- core -->
     <suppress checks="(NPathComplexity|ClassFanOutComplexity|CyclomaticComplexity|ClassDataAbstractionCoupling|FinalLocalVariable|LocalVariableName|MemberName|ParameterName|MethodLength|JavaNCSS|AvoidStarImport)"
               files="core[\\/]src[\\/](generated|generated-test)[\\/].+.java$"/>
-    <suppress checks="NPathComplexity" files="(ClusterTestExtensions|KafkaApisBuilder).java"/>
-    <suppress checks="MethodLength"
-              files="(KafkaClusterTestKit).java"/>
+    <suppress checks="MethodLength|NPathComplexity|CyclomaticComplexity|JavaNCSS" files="(ClusterTestExtensions|KafkaApisBuilder|KafkaClusterTestKit).java"/>
 
     <!-- Clients -->
     <suppress id="dontUseSystemExit"

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

@@ -22,22 +22,18 @@ import java.util
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantLock
 import java.util.concurrent.{CompletableFuture, ExecutionException, TimeUnit, TimeoutException}
-
 import kafka.cluster.Broker.ServerInfo
 import kafka.coordinator.group.GroupCoordinator
 import kafka.coordinator.transaction.{ProducerIdManager, TransactionCoordinator}
 import kafka.log.LogManager
 import kafka.network.{DataPlaneAcceptor, SocketServer}
-import kafka.raft.RaftManager
+import kafka.raft.KafkaRaftManager
 import kafka.security.CredentialProvider
-import kafka.server.KafkaRaftServer.ControllerRole
-import kafka.server.metadata.BrokerServerMetrics
-import kafka.server.metadata.{BrokerMetadataListener, BrokerMetadataPublisher, BrokerMetadataSnapshotter, ClientQuotaMetadataManager, KRaftMetadataCache, SnapshotWriterBuilder}
+import kafka.server.metadata.{BrokerMetadataPublisher, ClientQuotaMetadataManager, DynamicConfigPublisher, KRaftMetadataCache}
 import kafka.utils.{CoreUtils, KafkaScheduler}
 import org.apache.kafka.common.feature.SupportedVersionRange
 import org.apache.kafka.common.message.ApiMessageType.ListenerType
 import org.apache.kafka.common.message.BrokerRegistrationRequestData.{Listener, ListenerCollection}
-import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.security.scram.internals.ScramMechanism
@@ -46,45 +42,29 @@ import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time, Utils}
 import org.apache.kafka.common.{ClusterResource, Endpoint}
 import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer
 import org.apache.kafka.metadata.{BrokerState, VersionRange}
-import org.apache.kafka.raft.RaftConfig.AddressSpec
-import org.apache.kafka.raft.{RaftClient, RaftConfig}
+import org.apache.kafka.raft.RaftConfig
 import org.apache.kafka.server.authorizer.Authorizer
 import org.apache.kafka.server.common.ApiMessageAndVersion
-import org.apache.kafka.server.fault.FaultHandler
 import org.apache.kafka.server.metrics.KafkaYammerMetrics
-import org.apache.kafka.snapshot.SnapshotWriter
 
 import scala.collection.{Map, Seq}
-import scala.compat.java8.OptionConverters._
 import scala.jdk.CollectionConverters._
 
 
-class BrokerSnapshotWriterBuilder(raftClient: RaftClient[ApiMessageAndVersion])
-    extends SnapshotWriterBuilder {
-  override def build(committedOffset: Long,
-                     committedEpoch: Int,
-                     lastContainedLogTime: Long): Option[SnapshotWriter[ApiMessageAndVersion]] = {
-    raftClient.createSnapshot(committedOffset, committedEpoch, lastContainedLogTime).asScala
-  }
-}
-
 /**
  * A Kafka broker that runs in KRaft (Kafka Raft) mode.
  */
 class BrokerServer(
-  val config: KafkaConfig,
-  val metaProps: MetaProperties,
-  val raftManager: RaftManager[ApiMessageAndVersion],
-  val time: Time,
-  val metrics: Metrics,
-  val brokerMetrics: BrokerServerMetrics,
-  val threadNamePrefix: Option[String],
+  val jointServer: JointServer,
   val initialOfflineDirs: Seq[String],
-  val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]],
-  val fatalFaultHandler: FaultHandler,
-  val metadataLoadingFaultHandler: FaultHandler,
-  val metadataPublishingFaultHandler: FaultHandler
 ) extends KafkaBroker {
+  val threadNamePrefix = jointServer.threadNamePrefix
+  val config = jointServer.config
+  val time = jointServer.time
+  val metrics = jointServer.metrics
+  def raftManager: KafkaRaftManager[ApiMessageAndVersion] = jointServer.raftManager
+
+  class Builder
 
   override def brokerState: BrokerState = Option(lifecycleManager).
     flatMap(m => Some(m.state)).getOrElse(BrokerState.NOT_RUNNING)
@@ -144,11 +124,7 @@ class BrokerServer(
 
   @volatile var brokerTopicStats: BrokerTopicStats = _
 
-  val clusterId: String = metaProps.clusterId
-
-  var metadataSnapshotter: Option[BrokerMetadataSnapshotter] = None
-
-  var metadataListener: BrokerMetadataListener = _
+  val clusterId: String = jointServer.metaProps.clusterId
 
   var metadataPublisher: BrokerMetadataPublisher = _
 
@@ -180,9 +156,9 @@ class BrokerServer(
   override def startup(): Unit = {
     if (!maybeChangeStatus(SHUTDOWN, STARTING)) return
     try {
-      info("Starting broker")
+      jointServer.startForBroker()
 
-      config.dynamicConfig.initialize(zkClientOpt = None)
+      info("Starting broker")
 
       lifecycleManager = new BrokerLifecycleManager(config,
         time,
@@ -211,8 +187,8 @@ class BrokerServer(
       tokenCache = new DelegationTokenCache(ScramMechanism.mechanismNames)
       credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, tokenCache)
 
-      val controllerNodes = RaftConfig.voterConnectionsToNodes(controllerQuorumVotersFuture.get()).asScala
-      val controllerNodeProvider = RaftControllerNodeProvider(raftManager, config, controllerNodes)
+      val controllerNodes = RaftConfig.voterConnectionsToNodes(jointServer.controllerQuorumVotersFuture.get()).asScala
+      val controllerNodeProvider = RaftControllerNodeProvider(jointServer.raftManager, config, controllerNodes)
 
       clientToControllerChannelManager = BrokerToControllerChannelManager(
         controllerNodeProvider,
@@ -304,25 +280,6 @@ class BrokerServer(
         ConfigType.Topic -> new TopicConfigHandler(logManager, config, quotaManagers, None),
         ConfigType.Broker -> new BrokerConfigHandler(config, quotaManagers))
 
-      if (!config.processRoles.contains(ControllerRole)) {
-        // If no controller is defined, we rely on the broker to generate snapshots.
-        metadataSnapshotter = Some(new BrokerMetadataSnapshotter(
-          config.nodeId,
-          time,
-          threadNamePrefix,
-          new BrokerSnapshotWriterBuilder(raftManager.client)
-        ))
-      }
-
-      metadataListener = new BrokerMetadataListener(
-        config.nodeId,
-        time,
-        threadNamePrefix,
-        config.metadataSnapshotMaxNewRecordBytes,
-        metadataSnapshotter,
-        brokerMetrics,
-        metadataLoadingFaultHandler)
-
       val networkListeners = new ListenerCollection()
       config.effectiveAdvertisedListeners.foreach { ep =>
         networkListeners.add(new Listener().
@@ -347,16 +304,13 @@ class BrokerServer(
         config.brokerSessionTimeoutMs.toLong
       )
       lifecycleManager.start(
-        () => metadataListener.highestMetadataOffset,
+        () => jointServer.metadataLoader.lastAppliedOffset(),
         brokerLifecycleChannelManager,
-        metaProps.clusterId,
+        clusterId,
         networkListeners,
         featuresRemapped
       )
 
-      // Register a listener with the Raft layer to receive metadata event notifications
-      raftManager.register(metadataListener)
-
       val endpoints = new util.ArrayList[Endpoint](networkListeners.size())
       var interBrokerListener: Endpoint = null
       networkListeners.iterator().forEachRemaining(listener => {
@@ -426,30 +380,40 @@ class BrokerServer(
         DataPlaneAcceptor.ThreadPrefix)
 
       // Block until we've caught up with the latest metadata from the controller quorum.
-      lifecycleManager.initialCatchUpFuture.get()
+      info("Waiting for broker metadata to catch up.")
+      try {
+        lifecycleManager.initialCatchUpFuture.get(20, TimeUnit.SECONDS)
+      } catch {
+        case t: Throwable => throw new RuntimeException("Received a fatal error while " +
+          "waiting for the broker to catch up with the current cluster metadata.", t)
+      }
 
       // Apply the metadata log changes that we've accumulated.
+      val metadataPublishingFaultHandler = jointServer.metadataPublishingFaultHandler
       metadataPublisher = new BrokerMetadataPublisher(config,
         metadataCache,
         logManager,
         replicaManager,
         groupCoordinator,
         transactionCoordinator,
         clientQuotaMetadataManager,
-        dynamicConfigHandlers.toMap,
+        new DynamicConfigPublisher(
+          config,
+          metadataPublishingFaultHandler,
+          dynamicConfigHandlers.toMap),
         authorizer,
-        fatalFaultHandler,
+        jointServer.initialBrokerMetadataLoadFaultHandler,
         metadataPublishingFaultHandler)
 
       // Tell the metadata listener to start publishing its output, and wait for the first
       // 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.
       try {
-        metadataListener.startPublishing(metadataPublisher).get()
+        jointServer.metadataLoader.installPublishers(List(metadataPublisher).asJava).get(20, TimeUnit.SECONDS)
       } catch {
         case t: Throwable => throw new RuntimeException("Received a fatal error while " +
-          "waiting for the broker to catch up with the current cluster metadata.", t)
+          "publishing the initial metadata update.", t)
       }
 
       // Log static broker configurations.
@@ -470,7 +434,7 @@ 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.
       try {
-        lifecycleManager.setReadyToUnfence().get()
+        lifecycleManager.setReadyToUnfence().get(20, TimeUnit.SECONDS)
       } catch {
         case t: Throwable => throw new RuntimeException("Received a fatal error while " +
           "waiting for the broker to be unfenced.", t)
@@ -506,9 +470,6 @@ class BrokerServer(
         }
       }
 
-      if (metadataListener != null)
-        metadataListener.beginShutdown()
-
       lifecycleManager.beginShutdown()
 
       // Stop socket server to stop accepting any more connections and requests.
@@ -523,10 +484,6 @@ class BrokerServer(
       if (controlPlaneRequestProcessor != null)
         CoreUtils.swallow(controlPlaneRequestProcessor.close(), this)
       CoreUtils.swallow(authorizer.foreach(_.close()), this)
-      if (metadataListener != null) {
-        CoreUtils.swallow(metadataListener.close(), this)
-      }
-      metadataSnapshotter.foreach(snapshotter => CoreUtils.swallow(snapshotter.close(), this))
 
       /**
        * We must shutdown the scheduler early because otherwise, the scheduler could touch other
@@ -564,19 +521,18 @@ class BrokerServer(
       if (quotaManagers != null)
         CoreUtils.swallow(quotaManagers.shutdown(), this)
 
-      if (socketServer != null)
+      if (socketServer != null) {
         CoreUtils.swallow(socketServer.shutdown(), this)
+      }
       if (metrics != null)
         CoreUtils.swallow(metrics.close(), this)
       if (brokerTopicStats != null)
         CoreUtils.swallow(brokerTopicStats.close(), this)
 
-      // Clear all reconfigurable instances stored in DynamicBrokerConfig
-      config.dynamicConfig.clear()
-
       isShuttingDown.set(false)
 
       CoreUtils.swallow(lifecycleManager.close(), this)
+      jointServer.stopForBroker()
 
       CoreUtils.swallow(AppInfoParser.unregisterAppInfo(MetricsPrefix, config.nodeId.toString, metrics), this)
       info("shut down completed")