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

Code before:
      }
    }
    raftManager.replicatedLog.latestSnapshotId().ifPresent { latestSnapshotId =>
      raftManager.replicatedLog.readSnapshot(latestSnapshotId).ifPresent { rawSnapshotReader =>
        Using.resource(
          RecordsSnapshotReader.of(
            rawSnapshotReader,
            raftManager.recordSerde,
            BufferSupplier.create(),
            KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
            true,
            logContext
          )
        ) { reader =>
          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(dynamicDefaultConfigs, configRecord.name(), configRecord.value())
                    }
                  }
              }
            }
          }
          val configHandler = new BrokerConfigHandler(config, quotaManagers)
          configHandler.processConfigChanges("", dynamicDefaultConfigs)
          configHandler.processConfigChanges(config.brokerId.toString, dynamicDefaultConfigs)
        }
      }
    }
  }
}

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]()

Review comment (suggestion):
Hmm, this doesn't look right. If `configRecord.resourceName() == config.brokerId.toString`, it's a broker level config and it seems that we should use dynamicPerBrokerConfigs.

Code after:
      }
    }
    raftManager.replicatedLog.latestSnapshotId().ifPresent { latestSnapshotId =>
      raftManager.replicatedLog.readSnapshot(latestSnapshotId).ifPresent { rawSnapshotReader =>
        Using.resource(
          RecordsSnapshotReader.of(
            rawSnapshotReader,
            raftManager.recordSerde,
            BufferSupplier.create(),
            KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
            true,
            logContext
          )
        ) { reader =>
          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("", dynamicDefaultConfigs)
          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]()