Avoid duplicate config logging at startup
Kafka has duplicate configuration information log information printing during startup, repeated information printing will bring confusion to users.It is better to add log information before and after repeating the configuration information.
Reviewers: Guozhang Wang
@@ -212,7 +212,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
zkClientOpt.foreach { zkClient =>
val adminZkClient = new AdminZkClient(zkClient)
- updateDefaultConfig(adminZkClient.fetchEntityConfig(ConfigType.Broker, ConfigEntityName.Default))
+ updateDefaultConfig(adminZkClient.fetchEntityConfig(ConfigType.Broker, ConfigEntityName.Default), false)
val props = adminZkClient.fetchEntityConfig(ConfigType.Broker, kafkaConfig.brokerId.toString)
val brokerConfig = maybeReEncodePasswords(props, adminZkClient)
updateBrokerConfig(kafkaConfig.brokerId, brokerConfig)
@@ -291,23 +291,23 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
dynamicDefaultConfigs.clone()
}
- private[server] def updateBrokerConfig(brokerId: Int, persistentProps: Properties): Unit = CoreUtils.inWriteLock(lock) {
+ private[server] def updateBrokerConfig(brokerId: Int, persistentProps: Properties, doLog: Boolean = true): Unit = CoreUtils.inWriteLock(lock) {
try {
val props = fromPersistentProps(persistentProps, perBrokerConfig = true)
dynamicBrokerConfigs.clear()
dynamicBrokerConfigs ++= props.asScala
- updateCurrentConfig()
+ updateCurrentConfig(doLog)
} catch {
case e: Exception => error(s"Per-broker configs of $brokerId could not be applied: ${persistentProps.keys()}", e)
}
}
- private[server] def updateDefaultConfig(persistentProps: Properties): Unit = CoreUtils.inWriteLock(lock) {
+ private[server] def updateDefaultConfig(persistentProps: Properties, doLog: Boolean = true): Unit = CoreUtils.inWriteLock(lock) {
try {
val props = fromPersistentProps(persistentProps, perBrokerConfig = false)
dynamicDefaultConfigs.clear()
dynamicDefaultConfigs ++= props.asScala
- updateCurrentConfig()
+ updateCurrentConfig(doLog)
} catch {
case e: Exception => error(s"Cluster default configs could not be applied: ${persistentProps.keys()}", e)
}
@@ -514,14 +514,14 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
}
}
- private def updateCurrentConfig(): Unit = {
+ private def updateCurrentConfig(doLog: Boolean): Unit = {
val newProps = mutable.Map[String, String]()
newProps ++= staticBrokerConfigs
overrideProps(newProps, dynamicDefaultConfigs)
overrideProps(newProps, dynamicBrokerConfigs)
val oldConfig = currentConfig
- val (newConfig, brokerReconfigurablesToUpdate) = processReconfiguration(newProps, validateOnly = false)
+ val (newConfig, brokerReconfigurablesToUpdate) = processReconfiguration(newProps, validateOnly = false, doLog)
if (newConfig ne currentConfig) {
currentConfig = newConfig
kafkaConfig.updateCurrentConfig(newConfig)
@@ -531,8 +531,8 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
}
}
- private def processReconfiguration(newProps: Map[String, String], validateOnly: Boolean): (KafkaConfig, List[BrokerReconfigurable]) = {
- val newConfig = new KafkaConfig(newProps.asJava, !validateOnly, None)
+ private def processReconfiguration(newProps: Map[String, String], validateOnly: Boolean, doLog: Boolean = false): (KafkaConfig, List[BrokerReconfigurable]) = {
+ val newConfig = new KafkaConfig(newProps.asJava, doLog, None)
val (changeMap, deletedKeySet) = updatedConfigs(newConfig.originalsFromThisConfig, currentConfig.originals)
if (changeMap.nonEmpty || deletedKeySet.nonEmpty) {
try {
