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

Code before:

  /**
   * Clear all cached values. This is used to clear state on broker shutdown to avoid
   * exceptions in tests when broker is restarted. These fields are re-initialized when
   * broker starts up.
   */
  private[server] def clear(): Unit = {
    dynamicBrokerConfigs.clear()
    dynamicDefaultConfigs.clear()
    reconfigurables.clear()
    brokerReconfigurables.clear()
  }

  /**
   * Add reconfigurables to be notified when a dynamic broker config is updated.
   *
   * `Reconfigurable` is the public API used by configurable plugins like metrics reporter
   * and quota callbacks. These are reconfigured before `KafkaConfig` is updated so that
   * the update can be aborted if `reconfigure()` fails with an exception.
   *
   * `BrokerReconfigurable` is used for internal reconfigurable classes. These are
   * reconfigured after `KafkaConfig` is updated so that they can access `KafkaConfig`
   * directly. They are provided both old and new configs.
   */
  def addReconfigurables(kafkaServer: KafkaBroker): Unit = {
    kafkaServer.authorizer match {
      case Some(authz: Reconfigurable) => addReconfigurable(authz)
      case _ =>
    }
    addReconfigurable(kafkaServer.kafkaYammerMetrics)
    addReconfigurable(new DynamicMetricsReporters(kafkaConfig.brokerId, kafkaServer.config, kafkaServer.metrics, kafkaServer.clusterId))
    addReconfigurable(new DynamicClientQuotaCallback(kafkaServer.quotaManagers, kafkaServer.config))

    addBrokerReconfigurable(new BrokerDynamicThreadPool(kafkaServer))
    addBrokerReconfigurable(new DynamicLogConfig(kafkaServer.logManager))
    addBrokerReconfigurable(new DynamicListenerConfig(kafkaServer))
    addBrokerReconfigurable(kafkaServer.socketServer)
    addBrokerReconfigurable(new DynamicProducerStateManagerConfig(kafkaServer.logManager.producerStateManagerConfig))
    addBrokerReconfigurable(new DynamicRemoteLogConfig(kafkaServer))
  }

  /**
   * Add reconfigurables to be notified when a dynamic controller config is updated.
   */
  def addReconfigurables(controller: ControllerServer): Unit = {
    controller.authorizer match {
      case Some(authz: Reconfigurable) => addReconfigurable(authz)
      case _ =>
    }
    if (!kafkaConfig.processRoles.contains(ProcessRole.BrokerRole)) {
      // only add these if the controller isn't also running the broker role

Review comment (suggestion):
`dynamicPerBrokerConfigs` should be replaced by `dynamicDefaultConfigs`. I have opened https://issues.apache.org/jira/browse/KAFKA-19642 to fix it

Code after:
          rawSnapshotReader,
          raftManager.recordSerde,
          BufferSupplier.create(),
          KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
          true
        )
        val dynamicPerBrokerConfigs = new Properties()
        val dynamicDefaultConfigs = new Properties()
        while (reader.hasNext) {
          val batch = reader.next()
          batch.forEach(record => {
            if (record.message().apiKey() == MetadataRecordType.CONFIG_RECORD.id) {
              val configRecord = record.message().asInstanceOf[ConfigRecord]
              if (DynamicBrokerConfig.AllDynamicConfigs.contains(configRecord.name()) &&
                configRecord.resourceType() == ConfigResource.Type.BROKER.id()) {
                if (configRecord.resourceName().isEmpty) {
                  putOrRemoveIfNull(dynamicDefaultConfigs, configRecord.name(), configRecord.value())
                } else if (configRecord.resourceName() == config.brokerId.toString) {
                  putOrRemoveIfNull(dynamicPerBrokerConfigs, configRecord.name(), configRecord.value())
                }
              }
            }
          })
        }
        val configHandler = new BrokerConfigHandler(config, quotaManagers)
        configHandler.processConfigChanges("", dynamicPerBrokerConfigs)
        configHandler.processConfigChanges(config.brokerId.toString, dynamicPerBrokerConfigs)
      })
    })
  }
}

class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging {

  private[server] val staticBrokerConfigs = ConfigDef.convertToStringMapWithPasswordValues(kafkaConfig.originalsFromThisConfig).asScala
  private[server] val staticDefaultConfigs = ConfigDef.convertToStringMapWithPasswordValues(KafkaConfig.defaultValues.asJava).asScala
  private val dynamicBrokerConfigs = mutable.Map[String, String]()
  private val dynamicDefaultConfigs = mutable.Map[String, String]()

  // Use COWArrayList to prevent concurrent modification exception when an item is added by one thread to these
  // collections, while another thread is iterating over them.
  private[server] val reconfigurables = new CopyOnWriteArrayList[Reconfigurable]()
  private val brokerReconfigurables = new CopyOnWriteArrayList[BrokerReconfigurable]()
  private val lock = new ReentrantReadWriteLock
  private var metricsReceiverPluginOpt: Option[ClientMetricsReceiverPlugin] = _
  private var currentConfig: KafkaConfig = _

  private[server] def initialize(clientMetricsReceiverPluginOpt: Option[ClientMetricsReceiverPlugin]): Unit = {
    currentConfig = new KafkaConfig(kafkaConfig.props, false)
    metricsReceiverPluginOpt = clientMetricsReceiverPluginOpt
  }