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

Code before:
      }
    }
    props
  }

  private[server] def readDynamicBrokerConfigsFromSnapshot(
    raftManager: KafkaRaftManager[ApiMessageAndVersion],
    config: KafkaConfig,
    quotaManagers: QuotaFactory.QuotaManagers
  ): Unit = {
    raftManager.replicatedLog.latestSnapshotId().ifPresent(latestSnapshotId => {
      raftManager.replicatedLog.readSnapshot(latestSnapshotId).ifPresent(rawSnapshotReader => {
        val reader = RecordsSnapshotReader.of(
          rawSnapshotReader,
          raftManager.recordSerde,
          BufferSupplier.create(),
          KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
          true
        )
        val dynamicBrokerConfigs = 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())) {
                dynamicBrokerConfigs.put(configRecord.name(), configRecord.value())
              }
            }
          })
        }
        new BrokerConfigHandler(config, quotaManagers).processConfigChanges(config.brokerId.toString, dynamicBrokerConfigs)
      })
    })
  }
}

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 = _

Review comment (suggestion):
This is not quite right, because you aren't distinguishing between broker configurations and other kinds of configuration. You're also assuming that if you find a broker configuration, it applies to this node, which may not be the case.

Code after:
    config: KafkaConfig,
    quotaManagers: QuotaFactory.QuotaManagers
  ): Unit = {
    def putOrRemoveIfNull(props: Properties, key: String, value: String): Unit = {
      if (value == null) {
        props.remove(key)
      } else {
        props.put(key, value)
      }
    }
    raftManager.replicatedLog.latestSnapshotId().ifPresent(latestSnapshotId => {
      raftManager.replicatedLog.readSnapshot(latestSnapshotId).ifPresent(rawSnapshotReader => {
        val reader = RecordsSnapshotReader.of(
          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]()