Repository: apache/kafka
File: core/src/main/scala/kafka/server/SharedServer.scala (Scala)

Code before:
      Option(controllerServerMetrics).foreach(_.incrementMetadataErrorCount())
      // Note: snapshot generation does not need to be disabled for a publishing fault.
    })

  private def start(listenerEndpoints: Endpoints): Unit = synchronized {
    if (started) {
      debug("SharedServer has already been started.")
    } else {
      info("Starting SharedServer")
      try {
        if (metrics == null) {
          // Recreate the metrics object if we're restarting a stopped SharedServer object.
          // This is only done in tests.
          metrics = new Metrics()
        }
        sharedServerConfig.dynamicConfig.initialize(clientTelemetryExporterPluginOpt = None)

        if (sharedServerConfig.processRoles.contains(ProcessRole.BrokerRole)) {
          brokerMetrics = new BrokerServerMetrics(metrics)
        }
        if (sharedServerConfig.processRoles.contains(ProcessRole.ControllerRole)) {
          controllerServerMetrics = new ControllerMetadataMetrics(Optional.of(KafkaYammerMetrics.defaultRegistry()))
        }

        val externalKRaftMetrics: ExternalKRaftMetrics = ignoredStaticVoters => {
          if (brokerMetrics != null) brokerMetrics.setIgnoredStaticVoters(ignoredStaticVoters)
          if (controllerServerMetrics != null) controllerServerMetrics.setIgnoredStaticVoters(ignoredStaticVoters)
        }

        val _raftManager = new KafkaRaftManager[ApiMessageAndVersion](
          clusterId,
          sharedServerConfig,
          metaPropsEnsemble.logDirProps.get(metaPropsEnsemble.metadataLogDir.get).directoryId.get,
          new MetadataRecordSerde,
          KafkaRaftServer.MetadataPartition,
          KafkaRaftServer.MetadataTopicId,
          time,
          metrics,
          externalKRaftMetrics,
          Some(s"kafka-${sharedServerConfig.nodeId}-raft"), // No dash expected at the end
          controllerQuorumVotersFuture,
          bootstrapServers,
          listenerEndpoints,
          raftManagerFaultHandler
        )
        raftManager = _raftManager
        _raftManager.startup()

        nodeMetrics = new NodeMetrics(metrics, controllerConfig.unstableFeatureVersionsEnabled)
        metadataLoaderMetrics = if (brokerMetrics != null) {
          new MetadataLoaderMetrics(

Review comment (bug):
Would you mind creating a snapshot to avoid the race condition?
```scala
          Option(brokerMetrics).foreach(_.setIgnoredStaticVoters(ignoredStaticVoters))
          Option(controllerServerMetrics).foreach(_.setIgnoredStaticVoters(ignoredStaticVoters))
```

Code after:
      Option(controllerServerMetrics).foreach(_.incrementMetadataErrorCount())
      // Note: snapshot generation does not need to be disabled for a publishing fault.
    })

  private def start(listenerEndpoints: Endpoints): Unit = synchronized {
    if (started) {
      debug("SharedServer has already been started.")
    } else {
      info("Starting SharedServer")
      try {
        if (metrics == null) {
          // Recreate the metrics object if we're restarting a stopped SharedServer object.
          // This is only done in tests.
          metrics = new Metrics()
        }
        sharedServerConfig.dynamicConfig.initialize(clientTelemetryExporterPluginOpt = None)

        if (sharedServerConfig.processRoles.contains(ProcessRole.BrokerRole)) {
          brokerMetrics = new BrokerServerMetrics(metrics)
        }
        if (sharedServerConfig.processRoles.contains(ProcessRole.ControllerRole)) {
          controllerServerMetrics = new ControllerMetadataMetrics(Optional.of(KafkaYammerMetrics.defaultRegistry()))
        }

        val externalKRaftMetrics: ExternalKRaftMetrics = ignoredStaticVoters => {
          Option(brokerMetrics).foreach(_.setIgnoredStaticVoters(ignoredStaticVoters))
          Option(controllerServerMetrics).foreach(_.setIgnoredStaticVoters(ignoredStaticVoters))
        }

        val _raftManager = new KafkaRaftManager[ApiMessageAndVersion](
          clusterId,
          sharedServerConfig,
          metaPropsEnsemble.logDirProps.get(metaPropsEnsemble.metadataLogDir.get).directoryId.get,
          new MetadataRecordSerde,
          KafkaRaftServer.MetadataPartition,
          KafkaRaftServer.MetadataTopicId,
          time,
          metrics,
          externalKRaftMetrics,
          Some(s"kafka-${sharedServerConfig.nodeId}-raft"), // No dash expected at the end
          controllerQuorumVotersFuture,
          bootstrapServers,
          listenerEndpoints,
          raftManagerFaultHandler
        )
        raftManager = _raftManager
        _raftManager.startup()

        nodeMetrics = new NodeMetrics(metrics, controllerConfig.unstableFeatureVersionsEnabled)
        metadataLoaderMetrics = if (brokerMetrics != null) {
          new MetadataLoaderMetrics(