cmccabe commented on code in PR #12837:
URL: https://github.com/apache/kafka/pull/12837#discussion_r1022011553
##########
core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala:
##########
@@ -99,25 +94,28 @@ class KRaftQuorumImplementation(
startup: Boolean,
threadNamePrefix: Option[String],
): KafkaBroker = {
- val metrics = new Metrics()
- val broker = new BrokerServer(config = config,
- metaProps = new MetaProperties(clusterId, config.nodeId),
- raftManager = raftManager,
- time = time,
- metrics = metrics,
- brokerMetrics = BrokerServerMetrics(metrics),
- threadNamePrefix = Some("Broker%02d_".format(config.nodeId)),
- initialOfflineDirs = Seq(),
- controllerQuorumVotersFuture = controllerQuorumVotersFuture,
- fatalFaultHandler = faultHandler,
- metadataLoadingFaultHandler = faultHandler,
- metadataPublishingFaultHandler = faultHandler)
- if (startup) broker.startup()
- broker
+ val jointServer = new JointServer(config,
+ new MetaProperties(clusterId, config.nodeId),
+ Time.SYSTEM,
+ new Metrics(),
+ Option("Broker%02d_".format(config.nodeId)),
+ controllerQuorumVotersFuture,
+ faultHandlerFactory)
+ var broker: BrokerServer = null
+ try {
+ broker = new BrokerServer(jointServer,
+ initialOfflineDirs = Seq())
+ if (startup) broker.startup()
+ broker
+ } catch {
+ case e: Throwable => {
+ if (broker != null) CoreUtils.swallow(broker.shutdown(), log)
Review Comment:
Currently, if broker is null, then jointServer cannot have been started.
However, just to make it simple to understand what is going on, I'll add an
unconditional call to `jointServer#stopForBroker` here
--
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
