MINOR: extract jointly owned parts of BrokerServer and ControllerServer

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

@@ -27,6 +27,7 @@ import kafka.coordinator.group.{GroupCoordinator, GroupCoordinatorAdapter}
 import kafka.coordinator.transaction.{ProducerIdManager, TransactionCoordinator}
 import kafka.log.LogManager
 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, KRaftMetadataCache, SnapshotWriterBuilder}
@@ -66,14 +67,14 @@ class BrokerSnapshotWriterBuilder(raftClient: RaftClient[ApiMessageAndVersion])
  * A Kafka broker that runs in KRaft (Kafka Raft) mode.
  */
 class BrokerServer(
-  val jointServer: JointServer,
+  val sharedServer: SharedServer,
   val initialOfflineDirs: Seq[String],
 ) extends KafkaBroker {
-  val threadNamePrefix = jointServer.threadNamePrefix
-  val config = jointServer.config
-  val time = jointServer.time
-  val metrics = jointServer.metrics
-  val raftManager = jointServer.raftManager
+  val threadNamePrefix = sharedServer.threadNamePrefix
+  val config = sharedServer.config
+  val time = sharedServer.time
+  val metrics = sharedServer.metrics
+  def raftManager: KafkaRaftManager[ApiMessageAndVersion] = sharedServer.raftManager
 
   override def brokerState: BrokerState = Option(lifecycleManager).
     flatMap(m => Some(m.state)).getOrElse(BrokerState.NOT_RUNNING)
@@ -133,7 +134,7 @@ class BrokerServer(
 
   @volatile var brokerTopicStats: BrokerTopicStats = _
 
-  val clusterId: String = jointServer.metaProps.clusterId
+  val clusterId: String = sharedServer.metaProps.clusterId
 
   var metadataSnapshotter: Option[BrokerMetadataSnapshotter] = None
 
@@ -169,7 +170,7 @@ class BrokerServer(
   override def startup(): Unit = {
     if (!maybeChangeStatus(SHUTDOWN, STARTING)) return
     try {
-      jointServer.startForBroker()
+      sharedServer.startForBroker()
 
       info("Starting broker")
 
@@ -202,7 +203,7 @@ class BrokerServer(
       tokenCache = new DelegationTokenCache(ScramMechanism.mechanismNames)
       credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, tokenCache)
 
-      val controllerNodes = RaftConfig.voterConnectionsToNodes(jointServer.controllerQuorumVotersFuture.get()).asScala
+      val controllerNodes = RaftConfig.voterConnectionsToNodes(sharedServer.controllerQuorumVotersFuture.get()).asScala
       val controllerNodeProvider = RaftControllerNodeProvider(raftManager, config, controllerNodes)
 
       clientToControllerChannelManager = BrokerToControllerChannelManager(
@@ -311,8 +312,8 @@ class BrokerServer(
         threadNamePrefix,
         config.metadataSnapshotMaxNewRecordBytes,
         metadataSnapshotter,
-        jointServer.brokerMetrics,
-        jointServer.metadataLoaderFaultHandler)
+        sharedServer.brokerMetrics,
+        sharedServer.metadataLoaderFaultHandler)
 
       val networkListeners = new ListenerCollection()
       config.effectiveAdvertisedListeners.foreach { ep =>
@@ -340,7 +341,7 @@ class BrokerServer(
       lifecycleManager.start(
         () => metadataListener.highestMetadataOffset,
         brokerLifecycleChannelManager,
-        jointServer.metaProps.clusterId,
+        sharedServer.metaProps.clusterId,
         networkListeners,
         featuresRemapped
       )
@@ -430,8 +431,8 @@ class BrokerServer(
         clientQuotaMetadataManager,
         dynamicConfigHandlers.toMap,
         authorizer,
-        jointServer.initialBrokerMetadataLoadFaultHandler,
-        jointServer.metadataPublishingFaultHandler)
+        sharedServer.initialBrokerMetadataLoadFaultHandler,
+        sharedServer.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,
@@ -564,7 +565,7 @@ class BrokerServer(
       isShuttingDown.set(false)
 
       CoreUtils.swallow(lifecycleManager.close(), this)
-      jointServer.stopForBroker()
+      sharedServer.stopForBroker()
 
       CoreUtils.swallow(AppInfoParser.unregisterAppInfo(MetricsPrefix, config.nodeId.toString, metrics), this)
       info("shut down completed")

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

@@ -52,18 +52,18 @@ import scala.compat.java8.OptionConverters._
  * A Kafka controller that runs in KRaft (Kafka Raft) mode.
  */
 class ControllerServer(
-  val jointServer: JointServer,
+  val sharedServer: SharedServer,
   val configSchema: KafkaConfigSchema,
   val bootstrapMetadata: BootstrapMetadata,
 ) extends Logging with KafkaMetricsGroup {
 
   import kafka.server.Server._
 
-  val config = jointServer.config
-  val time = jointServer.time
-  val metrics = jointServer.metrics
-  val threadNamePrefix = jointServer.threadNamePrefix.getOrElse("")
-  def raftManager: KafkaRaftManager[ApiMessageAndVersion] = jointServer.raftManager
+  val config = sharedServer.config
+  val time = sharedServer.time
+  val metrics = sharedServer.metrics
+  val threadNamePrefix = sharedServer.threadNamePrefix.getOrElse("")
+  def raftManager: KafkaRaftManager[ApiMessageAndVersion] = sharedServer.raftManager
 
   config.dynamicConfig.initialize(zkClientOpt = None)
 
@@ -103,7 +103,7 @@ class ControllerServer(
     new DynamicMetricReporterState(config.nodeId, config, metrics, clusterId)
   }
 
-  def clusterId: String = jointServer.metaProps.clusterId
+  def clusterId: String = sharedServer.metaProps.clusterId
 
   def startup(): Unit = {
     if (!maybeChangeStatus(SHUTDOWN, STARTING)) return
@@ -163,16 +163,16 @@ class ControllerServer(
         throw new ConfigException("No controller.listener.names defined for controller")
       }
 
-      jointServer.startForController()
+      sharedServer.startForController()
 
       createTopicPolicy = Option(config.
         getConfiguredInstance(CreateTopicPolicyClassNameProp, classOf[CreateTopicPolicy]))
       alterConfigPolicy = Option(config.
         getConfiguredInstance(AlterConfigPolicyClassNameProp, classOf[AlterConfigPolicy]))
 
-      val controllerNodes = RaftConfig.voterConnectionsToNodes(jointServer.controllerQuorumVotersFuture.get())
+      val controllerNodes = RaftConfig.voterConnectionsToNodes(sharedServer.controllerQuorumVotersFuture.get())
       val quorumFeatures = QuorumFeatures.create(config.nodeId,
-        jointServer.raftManager.apiVersions,
+        sharedServer.raftManager.apiVersions,
         QuorumFeatures.defaultFeatureMap(),
         controllerNodes)
 
@@ -185,7 +185,7 @@ class ControllerServer(
 
         val maxIdleIntervalNs = config.metadataMaxIdleIntervalNs.fold(OptionalLong.empty)(OptionalLong.of)
 
-        new QuorumController.Builder(config.nodeId, jointServer.metaProps.clusterId).
+        new QuorumController.Builder(config.nodeId, sharedServer.metaProps.clusterId).
           setTime(time).
           setThreadNamePrefix(threadNamePrefix).
           setConfigSchema(configSchema).
@@ -199,13 +199,13 @@ class ControllerServer(
           setSnapshotMaxIntervalMs(config.metadataSnapshotMaxIntervalMs).
           setLeaderImbalanceCheckIntervalNs(leaderImbalanceCheckIntervalNs).
           setMaxIdleIntervalNs(maxIdleIntervalNs).
-          setMetrics(jointServer.controllerMetrics).
+          setMetrics(sharedServer.controllerMetrics).
           setCreateTopicPolicy(createTopicPolicy.asJava).
           setAlterConfigPolicy(alterConfigPolicy.asJava).
           setConfigurationValidator(new ControllerConfigurationValidator()).
           setStaticConfig(config.originals).
           setBootstrapMetadata(bootstrapMetadata).
-          setFatalFaultHandler(jointServer.quorumControllerFaultHandler)
+          setFatalFaultHandler(sharedServer.quorumControllerFaultHandler)
       }
       authorizer match {
         case Some(a: ClusterMetadataAuthorizer) => controllerBuilder.setAuthorizer(a)
@@ -229,7 +229,7 @@ class ControllerServer(
         controller,
         raftManager,
         config,
-        jointServer.metaProps,
+        sharedServer.metaProps,
         controllerNodes.asScala.toSeq,
         apiVersionManager)
       controllerApisHandlerPool = new KafkaRequestHandlerPool(config.nodeId,
@@ -265,7 +265,7 @@ class ControllerServer(
       info("shutting down")
       // Ensure that we're not the Raft leader prior to shutting down our socket server, for a
       // smoother transition.
-      jointServer.ensureNotRaftLeader()
+      sharedServer.ensureNotRaftLeader()
       if (socketServer != null)
         CoreUtils.swallow(socketServer.stopProcessingRequests(), this)
       if (controller != null)
@@ -284,7 +284,7 @@ class ControllerServer(
       createTopicPolicy.foreach(policy => CoreUtils.swallow(policy.close(), this))
       alterConfigPolicy.foreach(policy => CoreUtils.swallow(policy.close(), this))
       socketServerFirstBoundPortFuture.completeExceptionally(new RuntimeException("shutting down"))
-      jointServer.stopForController()
+      sharedServer.stopForController()
     } catch {
       case e: Throwable =>
         fatal("Fatal error during controller shutdown.", e)

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

@@ -64,7 +64,7 @@ class KafkaRaftServer(
   private val controllerQuorumVotersFuture = CompletableFuture.completedFuture(
     RaftConfig.parseVoterConnections(config.quorumVoters))
 
-  private val jointServer = new JointServer(
+  private val sharedServer = new SharedServer(
     config,
     metaProps,
     time,
@@ -76,7 +76,7 @@ class KafkaRaftServer(
 
   private val broker: Option[BrokerServer] = if (config.processRoles.contains(BrokerRole)) {
     Some(new BrokerServer(
-      jointServer,
+      sharedServer,
       offlineDirs
     ))
   } else {
@@ -85,7 +85,7 @@ class KafkaRaftServer(
 
   private val controller: Option[ControllerServer] = if (config.processRoles.contains(ControllerRole)) {
     Some(new ControllerServer(
-      jointServer,
+      sharedServer,
       KafkaRaftServer.configSchema,
       bootstrapMetadata,
     ))

core/src/main/scala/kafka/server/JointServer.scala → core/src/main/scala/kafka/server/SharedServer.scala

@@ -64,20 +64,20 @@ class StandardFaultHandlerFactory extends FaultHandlerFactory {
 }
 
 /**
- * The JointServer manages the components which are shared between the BrokerServer and
+ * The SharedServer manages the components which are shared between the BrokerServer and
  * ControllerServer. These shared components include the Raft manager, snapshot generator,
  * and metadata loader. A KRaft server running in combined mode as both a broker and a controller
- * will still contain only a single JointServer instance.
+ * will still contain only a single SharedServer instance.
  *
- * The JointServer will be started as soon as either the broker or the controller needs it,
+ * The SharedServer will be started as soon as either the broker or the controller needs it,
  * via the appropriate function (startForBroker or startForController). Similarly, it will be
  * stopped as soon as neither the broker nor the controller need it, via stopForBroker or
  * stopForController. One way of thinking about this is that both the broker and the controller
  * could hold a "reference" to this class, and we don't truly stop it until both have dropped
  * their reference. We opted to use two booleans here rather than a reference count in order to
  * make debugging easier and reduce the chance of resource leaks.
  */
-class JointServer(
+class SharedServer(
   val config: KafkaConfig,
   val metaProps: MetaProperties,
   val time: Time,
@@ -86,7 +86,7 @@ class JointServer(
   val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]],
   val faultHandlerFactory: FaultHandlerFactory
 ) extends Logging {
-  private val logContext: LogContext = new LogContext(s"[JointServer id=${config.nodeId}] ")
+  private val logContext: LogContext = new LogContext(s"[SharedServer id=${config.nodeId}] ")
   this.logIdent = logContext.logPrefix
   val stopped = new AtomicBoolean(false)
 
@@ -179,9 +179,9 @@ class JointServer(
     })
 
   private def start(): Unit = synchronized {
-    info("Starting JointServer")
+    info("Starting SharedServer")
     if (stopped.get()) {
-      throw new RuntimeException("Cannot restart stopped JointServer.")
+      throw new RuntimeException("Cannot restart stopped SharedServer.")
     }
     try {
       config.dynamicConfig.initialize(zkClientOpt = None)
@@ -203,10 +203,10 @@ class JointServer(
         threadNamePrefix,
         controllerQuorumVotersFuture)
       raftManager.startup()
-      debug("Completed JointServer startup.")
+      debug("Completed SharedServer startup.")
     } catch {
       case e: Throwable => {
-        error("Got exception while starting JointServer", e)
+        error("Got exception while starting SharedServer", e)
         stop()
       }
     }
@@ -223,9 +223,9 @@ class JointServer(
 
   private def stop(): Unit = {
     if (stopped.getAndSet(true)) {
-      debug("JointServer is already stopped")
+      debug("SharedServer is already stopped")
     } else {
-      info("Stopping JointServer")
+      info("Stopping SharedServer")
       if (raftManager != null) {
         CoreUtils.swallow(raftManager.shutdown(), this)
         raftManager = null

core/src/test/java/kafka/testkit/KafkaClusterTestKit.java

@@ -21,7 +21,7 @@
 import kafka.server.BrokerServer;
 import kafka.server.ControllerServer;
 import kafka.server.FaultHandlerFactory;
-import kafka.server.JointServer;
+import kafka.server.SharedServer;
 import kafka.server.KafkaConfig;
 import kafka.server.KafkaConfig$;
 import kafka.server.KafkaRaftServer;
@@ -199,7 +199,7 @@ private KafkaConfig createNodeConfig(TestKitNode node) {
         public KafkaClusterTestKit build() throws Exception {
             Map<Integer, ControllerServer> controllers = new HashMap<>();
             Map<Integer, BrokerServer> brokers = new HashMap<>();
-            Map<Integer, JointServer> jointServers = new HashMap<>();
+            Map<Integer, SharedServer> jointServers = new HashMap<>();
             /*
               Number of threads = Total number of brokers + Total number of controllers + Total number of Raft Managers
                                 = Total number of brokers + Total number of controllers * 2
@@ -223,7 +223,7 @@ public KafkaClusterTestKit build() throws Exception {
                     String threadNamePrefix = (nodes.brokerNodes().containsKey(node.id())) ?
                             String.format("colocated%d", node.id()) :
                             String.format("controller%d", node.id());
-                    JointServer jointServer = new JointServer(createNodeConfig(node),
+                    SharedServer sharedServer = new SharedServer(createNodeConfig(node),
                             MetaProperties.apply(nodes.clusterId().toString(), node.id()),
                             Time.SYSTEM,
                             new Metrics(),
@@ -233,12 +233,12 @@ public KafkaClusterTestKit build() throws Exception {
                     ControllerServer controller = null;
                     try {
                         controller = new ControllerServer(
-                                jointServer,
+                                sharedServer,
                                 KafkaRaftServer.configSchema(),
                                 bootstrapMetadata);
                     } catch (Throwable e) {
                         log.error("Error creating controller {}", node.id(), e);
-                        Utils.swallow(log, "jointServer.stopForController", () -> jointServer.stopForController());
+                        Utils.swallow(log, "sharedServer.stopForController", () -> sharedServer.stopForController());
                         if (controller != null) controller.shutdown();
                         throw e;
                     }
@@ -250,11 +250,11 @@ public KafkaClusterTestKit build() throws Exception {
                             connectFutureManager.registerPort(node.id(), port);
                         }
                     });
-                    jointServers.put(node.id(), jointServer);
+                    jointServers.put(node.id(), sharedServer);
                 }
                 for (BrokerNode node : nodes.brokerNodes().values()) {
-                    JointServer jointServer = jointServers.computeIfAbsent(node.id(),
-                        id -> new JointServer(createNodeConfig(node),
+                    SharedServer sharedServer = jointServers.computeIfAbsent(node.id(),
+                        id -> new SharedServer(createNodeConfig(node),
                             MetaProperties.apply(nodes.clusterId().toString(), id),
                             Time.SYSTEM,
                             new Metrics(),
@@ -264,11 +264,11 @@ public KafkaClusterTestKit build() throws Exception {
                     BrokerServer broker = null;
                     try {
                         broker = new BrokerServer(
-                                jointServer,
+                                sharedServer,
                                 JavaConverters.asScalaBuffer(Collections.<String>emptyList()).toSeq());
                     } catch (Throwable e) {
                         log.error("Error creating broker {}", node.id(), e);
-                        Utils.swallow(log, "jointServer.stopForBroker", () -> jointServer.stopForBroker());
+                        Utils.swallow(log, "sharedServer.stopForBroker", () -> sharedServer.stopForBroker());
                         if (broker != null) broker.shutdown();
                         throw e;
                     }
@@ -503,11 +503,11 @@ public Map<Integer, BrokerServer> brokers() {
     public Map<Integer, KafkaRaftManager<ApiMessageAndVersion>> raftManagers() {
         Map<Integer, KafkaRaftManager<ApiMessageAndVersion>> results = new HashMap<>();
         for (BrokerServer brokerServer : brokers().values()) {
-            results.put(brokerServer.config().brokerId(), brokerServer.jointServer().raftManager());
+            results.put(brokerServer.config().brokerId(), brokerServer.sharedServer().raftManager());
         }
         for (ControllerServer controllerServer : controllers().values()) {
             if (!results.containsKey(controllerServer.config().nodeId())) {
-                results.put(controllerServer.config().nodeId(), controllerServer.jointServer().raftManager());
+                results.put(controllerServer.config().nodeId(), controllerServer.sharedServer().raftManager());
             }
         }
         return results;