cmccabe commented on code in PR #12837:
URL: https://github.com/apache/kafka/pull/12837#discussion_r1022016615
##########
core/src/main/scala/kafka/server/KafkaRaftServer.scala:
##########
@@ -69,95 +64,56 @@ class KafkaRaftServer(
private val controllerQuorumVotersFuture = CompletableFuture.completedFuture(
RaftConfig.parseVoterConnections(config.quorumVoters))
- private val raftManager = new KafkaRaftManager[ApiMessageAndVersion](
- metaProps,
+ private val jointServer = new JointServer(
config,
- new MetadataRecordSerde,
- KafkaRaftServer.MetadataPartition,
- KafkaRaftServer.MetadataTopicId,
+ metaProps,
time,
metrics,
threadNamePrefix,
- controllerQuorumVotersFuture
+ controllerQuorumVotersFuture,
+ new StandardFaultHandlerFactory(),
)
private val broker: Option[BrokerServer] = if
(config.processRoles.contains(BrokerRole)) {
- val brokerMetrics = BrokerServerMetrics(metrics)
- val fatalFaultHandler = new ProcessExitingFaultHandler()
- val metadataLoadingFaultHandler = new LoggingFaultHandler("metadata
loading",
- () => brokerMetrics.metadataLoadErrorCount.getAndIncrement())
- val metadataApplyingFaultHandler = new LoggingFaultHandler("metadata
application",
- () => brokerMetrics.metadataApplyErrorCount.getAndIncrement())
Some(new BrokerServer(
- config,
- metaProps,
- raftManager,
- time,
- metrics,
- brokerMetrics,
- threadNamePrefix,
- offlineDirs,
- controllerQuorumVotersFuture,
- fatalFaultHandler,
- metadataLoadingFaultHandler,
- metadataApplyingFaultHandler
+ jointServer,
+ offlineDirs
))
} else {
None
}
private val controller: Option[ControllerServer] = if
(config.processRoles.contains(ControllerRole)) {
- val controllerMetrics = new
QuorumControllerMetrics(KafkaYammerMetrics.defaultRegistry(), time)
- val metadataFaultHandler = new LoggingFaultHandler("controller metadata",
- () => controllerMetrics.incrementMetadataErrorCount())
- val fatalFaultHandler = new ProcessExitingFaultHandler()
Some(new ControllerServer(
- metaProps,
- config,
- raftManager,
- time,
- metrics,
- controllerMetrics,
- threadNamePrefix,
- controllerQuorumVotersFuture,
+ jointServer,
KafkaRaftServer.configSchema,
- raftManager.apiVersions,
bootstrapMetadata,
- metadataFaultHandler,
- fatalFaultHandler
))
} else {
None
}
override def startup(): Unit = {
Mx4jLoader.maybeLoad()
- // Note that we startup `RaftManager` first so that the controller and
broker
- // can register listeners during initialization.
- raftManager.startup()
controller.foreach(_.startup())
broker.foreach(_.startup())
AppInfoParser.registerAppInfo(Server.MetricsPrefix,
config.brokerId.toString, metrics, time.milliseconds())
info(KafkaBroker.STARTED_MESSAGE)
}
override def shutdown(): Unit = {
+ // In combined mode, we want to shut down the broker first, since it may
take longer to complete.
+ // Additionally, the controller shutdown process stops the raft client
early on (currently),
+ // which would disrupt broker shutdown.
broker.foreach(_.shutdown())
- // The order of shutdown for `RaftManager` and `ControllerServer` is
backwards
- // compared to `startup()`. This is because the `SocketServer`
implementation that
- // we rely on to receive requests is owned by `ControllerServer`, so we
need it
- // to stick around until graceful shutdown of `RaftManager` can be
completed.
- raftManager.shutdown()
controller.foreach(_.shutdown())
CoreUtils.swallow(AppInfoParser.unregisterAppInfo(Server.MetricsPrefix,
config.brokerId.toString, metrics), this)
Review Comment:
I will add this to `JointServer#stop`.
It was previously being done in `BrokerServer#shutdown`, which handles
broker-only and combined mode. But I guess in the case of testing
`ControllerServer`, we leaked this resource.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
