You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cmccabe (via GitHub)" <gi...@apache.org> on 2024/09/23 23:14:04 UTC
[PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
cmccabe opened a new pull request, #17258:
URL: https://github.com/apache/kafka/pull/17258
Several Kafka log configurations in have synonyms. For example, log retention can be configured either by log.retention.ms, or by log.retention.minutes, or by log.retention.hours. There is also a faculty in Kafka to dynamically change broker configurations without restarting the broker. These dynamically set configurations are stored in the metadata log and override what is in the broker properties file.
Unfortunately, these two features interacted poorly; there was a bug where the dynamic log configuration update code ignored synonyms. For example, if you set log.retention.minutes and then reconfigured something unrelated that triggered the LogConfig update path, the retention value that you had configured was overwritten.
The reason for this was incorrect handling of synonyms. The code tried to treat the Kafka broker configuration as a bag of key/value entities rather than extracting the correct retention time (or other setting with overrides) from the KafkaConfig object.
Separately from the above bug, the code did not honor the value of dynamically configured synonyms: setting log.retention.minutes had no effect; only log.retention.ms was honored.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1776102833
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +744,14 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ val originalLogConfigMap = originalLogConfig.originals()
+ DynamicLogConfig.NonReconfigrableLogConfigs.foreach(k => {
+ Option(originalLogConfigMap.get(k)) match {
+ case None => newBrokerDefaults.remove(k)
+ case Some(v) => newBrokerDefaults.put(k, v)
}
- }
+ })
Review Comment:
> newBrokerDefaults has the config name with log prefix, right
no. it comes from `KafkaConfig.extractLogConfigMap`, which creates a map of log (aka topic) configurations, not broker configurations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1776101826
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +744,14 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ val originalLogConfigMap = originalLogConfig.originals()
+ DynamicLogConfig.NonReconfigrableLogConfigs.foreach(k => {
+ Option(originalLogConfigMap.get(k)) match {
Review Comment:
Non-reconfigurable configs are copied over from the *current* (not new) configuration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon merged PR #17258:
URL: https://github.com/apache/kafka/pull/17258
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1775989476
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +732,12 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ originalLogConfig.originals().forEach((k, v) => {
+ if (!DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
+ newBrokerDefaults.put(k, v)
Review Comment:
1. Do we need to translate the name from KafkaConfig to LogConfig as the original code does?
2. We started `newBrokerDefaults` with `newConfig.extractLogConfigMap` that includes non-reconfigurable configs. Where is the logic to remove them?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2371956905
For now, I'm going to change it back to the current behavior where we ignore the value of dynamically configured synonyms such as `log.retention.minutes`, etc. We can make that change in a separate PR if people feel strongly about it. (The fact that it's been like this since 3.0 seems like a good reason to leave it for now....)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1772706976
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -662,12 +662,27 @@ trait BrokerReconfigurable {
}
object DynamicLogConfig {
- // Exclude message.format.version for now since we need to check that the version
- // is supported on all brokers in the cluster.
- val ReconfigurableConfigs: Set[String] =
- ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.values.asScala.toSet - ServerLogConfigs.LOG_MESSAGE_FORMAT_VERSION_CONFIG
- val KafkaConfigToLogConfigName: Map[String, String] =
- ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.asScala.map { case (k, v) => (v, k) }
+ val ReconfigurableConfigs: Set[String] = {
+ val results = new util.HashSet[String]
+ ServerTopicConfigSynonyms.ALL_TOPIC_CONFIG_SYNONYMS.values().forEach(v =>
+ v.forEach(configSynonym => results.add(configSynonym.name())))
+
+ // Exclude message.format.version for now since we need to check that the version
+ // is supported on all brokers in the cluster.
+ results.remove(ServerLogConfigs.LOG_MESSAGE_FORMAT_VERSION_CONFIG)
+
+ results.asScala
+ }
+
+ val KafkaConfigToLogConfigName: Map[String, String] = {
Review Comment:
nit: After this PR, this variable is only used in `DynamicBrokerReconfigurationTest`. We can move it there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2371267476
Looks like there is a failing test:
> DynamicBrokerReconfigurationTest > "testConfigDescribeUsingAdminClient(String)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1776088538
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +744,14 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ val originalLogConfigMap = originalLogConfig.originals()
+ DynamicLogConfig.NonReconfigrableLogConfigs.foreach(k => {
+ Option(originalLogConfigMap.get(k)) match {
+ case None => newBrokerDefaults.remove(k)
+ case Some(v) => newBrokerDefaults.put(k, v)
Review Comment:
Not sure that I follow here. Why are we putting a non-reconfigurable config to newBrokerDefaults?
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +744,14 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ val originalLogConfigMap = originalLogConfig.originals()
+ DynamicLogConfig.NonReconfigrableLogConfigs.foreach(k => {
+ Option(originalLogConfigMap.get(k)) match {
+ case None => newBrokerDefaults.remove(k)
+ case Some(v) => newBrokerDefaults.put(k, v)
}
- }
+ })
Review Comment:
`newBrokerDefaults` has the config name with log prefix, right? Should we translate the name from `KafkaConfig` to `LogConfig` as the original code does?
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +744,14 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ val originalLogConfigMap = originalLogConfig.originals()
+ DynamicLogConfig.NonReconfigrableLogConfigs.foreach(k => {
+ Option(originalLogConfigMap.get(k)) match {
Review Comment:
Hmm, `originalLogConfigMap` is the current config. We should check from `newConfig`, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1775017624
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +732,7 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
Review Comment:
It has been a long time, so I may not remember correctly. But I think we used to only update configs which are `ReconfigurableConfigs` even if you updated ZooKeeper directly with new configs that included others. Non-reconfigurable configs would be picked up only on the next broker restart. Do we think it is safe to include all configs now either because we don't allow other configs to be updated in KRaft or because all log configs are reconfigurable now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2373050355
This issue was first reported in KAFKA-15266 and a PR #14119 is opened for it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "fvaleri (via GitHub)" <gi...@apache.org>.
fvaleri commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2370642153
> The log.retention.hours should be not a read-only config. It can be dynamically changed, as well as log.roll.hours.
I think we should also update the documentation, as they are both reported as read-only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "jayteej (via GitHub)" <gi...@apache.org>.
jayteej commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1772947639
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -662,12 +662,27 @@ trait BrokerReconfigurable {
}
object DynamicLogConfig {
- // Exclude message.format.version for now since we need to check that the version
- // is supported on all brokers in the cluster.
- val ReconfigurableConfigs: Set[String] =
- ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.values.asScala.toSet - ServerLogConfigs.LOG_MESSAGE_FORMAT_VERSION_CONFIG
- val KafkaConfigToLogConfigName: Map[String, String] =
- ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.asScala.map { case (k, v) => (v, k) }
+ val ReconfigurableConfigs: Set[String] = {
Review Comment:
nit (immutable style):
```
val ReconfigurableConfigs: Set[String] = {
ServerTopicConfigSynonyms.ALL_TOPIC_CONFIG_SYNONYMS
.values()
.asScala
.flatMap(v => v.asScala.map(configSynonym => configSynonym.name()))
.filterNot(_ == ServerLogConfigs.LOG_MESSAGE_FORMAT_VERSION_CONFIG)
.toSet
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "amangandhi94 (via GitHub)" <gi...@apache.org>.
amangandhi94 commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1774597780
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +732,7 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
Review Comment:
Just curious - Do we know why this logic was even added? I cant seem to understand if there was any benefit to doing this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2370685939
> > The log.retention.hours should be not a read-only config. It can be dynamically changed, as well as log.roll.hours.
>
> I think we should also update the documentation, as they are both reported as read-only.
Good point! But the doc is actually generated, and I just confirmed, after this PR, the doc will be correctly updated:
```
log.retention.hours
The number of hours to keep a log file before deleting it (in hours), tertiary to log.retention.ms property
Type: int
Default: 168
Valid Values:
Importance: high
Update Mode: cluster-wide
log.retention.minutes
The number of minutes to keep a log file before deleting it (in minutes), secondary to log.retention.ms property. If not set, the value in log.retention.hours is used
Type: int
Default: null
Valid Values:
Importance: high
Update Mode: cluster-wide
log.retention.ms
The number of milliseconds to keep a log file before deleting it (in milliseconds), If not set, the value in log.retention.minutes is used. If set to -1, no time limit is applied.
Type: long
Default: null
Valid Values:
Importance: high
Update Mode: cluster-wide
log.roll.hours
The maximum time before a new log segment is rolled out (in hours), secondary to log.roll.ms property
Type: int
Default: 168
Valid Values: [1,...]
Importance: high
Update Mode: cluster-wide
log.roll.jitter.hours
The maximum jitter to subtract from logRollTimeMillis (in hours), secondary to log.roll.jitter.ms property
Type: int
Default: 0
Valid Values: [0,...]
Importance: high
Update Mode: cluster-wide
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "jayteej (via GitHub)" <gi...@apache.org>.
jayteej commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1772948566
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -662,12 +662,27 @@ trait BrokerReconfigurable {
}
object DynamicLogConfig {
- // Exclude message.format.version for now since we need to check that the version
- // is supported on all brokers in the cluster.
- val ReconfigurableConfigs: Set[String] =
- ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.values.asScala.toSet - ServerLogConfigs.LOG_MESSAGE_FORMAT_VERSION_CONFIG
- val KafkaConfigToLogConfigName: Map[String, String] =
- ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.asScala.map { case (k, v) => (v, k) }
+ val ReconfigurableConfigs: Set[String] = {
+ val results = new util.HashSet[String]
+ ServerTopicConfigSynonyms.ALL_TOPIC_CONFIG_SYNONYMS.values().forEach(v =>
+ v.forEach(configSynonym => results.add(configSynonym.name())))
+
+ // Exclude message.format.version for now since we need to check that the version
+ // is supported on all brokers in the cluster.
+ results.remove(ServerLogConfigs.LOG_MESSAGE_FORMAT_VERSION_CONFIG)
+
+ results.asScala
+ }
+
+ val KafkaConfigToLogConfigName: Map[String, String] = {
+ val results = new util.HashMap[String, String]
+ ServerTopicConfigSynonyms.ALL_TOPIC_CONFIG_SYNONYMS.entrySet().forEach(e => {
Review Comment:
nit: As per above, use flatmap and toSet
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1775665743
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +732,7 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
Review Comment:
Thanks for this comment, @rajinisivaram . I think you're right that we should be excluding changes to configurations that don't appear in `DynamicLogConfig.ReconfigurableConfigs`. I have added some code to do this now. I think from a practical point of view, this only affects `message.format.version`, which we explicitly excluded from being dynamically reconfigurable.
> Do we think it is safe to include all configs now either because we don't allow other configs to be updated in KRaft or because all log configs are reconfigurable now?
KRaft has the same behavior as ZK, in that it lets you set any broker configuration you want, whether it's valid or not. There were too many people depending on this for us to change it in 3.x. Maybe in the future if someone creates a KIP...
Of course the command-line tool does its own validation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2376012733
All tests passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "fvaleri (via GitHub)" <gi...@apache.org>.
fvaleri commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2373258542
> Just wondering how can I get more attention to bugs or PR i have raised next time onwards?
In this case raising the bug as a blocker, due to the potential data loss. In general, sending an email to the dev mailing list would be enough to catch some more attention in my experience. Hope it helps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1776102102
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +744,14 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ val originalLogConfigMap = originalLogConfig.originals()
+ DynamicLogConfig.NonReconfigrableLogConfigs.foreach(k => {
+ Option(originalLogConfigMap.get(k)) match {
+ case None => newBrokerDefaults.remove(k)
+ case Some(v) => newBrokerDefaults.put(k, v)
Review Comment:
Because we want the non-reconfigurable configuration to have the same value (or lack of value) that it had previously. It should not change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "fvaleri (via GitHub)" <gi...@apache.org>.
fvaleri commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2370756685
> we publicly call it out in some form
Agreed. This is what happens right now, so it would be a behavior change. I think a simple release note would be fine.
```sh
$ bin/kafka-configs.sh --bootstrap-server :9092 --entity-type brokers --entity-name 2 --alter --add-config log.retention.minutes=1
Error while executing config command with args '--bootstrap-server :9092 --entity-type brokers --entity-name 2 --alter --add-config log.retention.minutes=1'
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.InvalidRequestException: Cannot update these configs dynamically: Set(log.retention.minutes)
at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2096)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:180)
at kafka.admin.ConfigCommand$.alterConfig(ConfigCommand.scala:390)
at kafka.admin.ConfigCommand$.processCommand(ConfigCommand.scala:351)
at kafka.admin.ConfigCommand$.main(ConfigCommand.scala:100)
at kafka.admin.ConfigCommand.main(ConfigCommand.scala)
Caused by: org.apache.kafka.common.errors.InvalidRequestException: Cannot update these configs dynamically: Set(log.retention.minutes)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1775717911
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +732,7 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
Review Comment:
See discussion here https://github.com/apache/kafka/pull/17258#discussion_r1775665743
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #17258:
URL: https://github.com/apache/kafka/pull/17258#issuecomment-2370816167
I've sent a mail dev/user mailing list to discuss it [here](https://lists.apache.org/thread/pn52rf3hrlz2fowftpfgp27o7o34h2kp). FYI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1776066292
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +732,12 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
- DynamicLogConfig.KafkaConfigToLogConfigName.get(k).foreach { configName =>
- if (v == null)
- newBrokerDefaults.remove(configName)
- else
- newBrokerDefaults.put(configName, v.asInstanceOf[AnyRef])
- }
+ val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
+ originalLogConfig.originals().forEach((k, v) => {
+ if (!DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
+ newBrokerDefaults.put(k, v)
Review Comment:
Let me fix this a bit. It should be sufficient to just remove the non-reconfigurable configs (really, singular config)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17584: Fix incorrect synonym handling for dynamic log configurations [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #17258:
URL: https://github.com/apache/kafka/pull/17258#discussion_r1776431850
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -732,17 +732,7 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaBroker) extends Brok
override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
val originalLogConfig = logManager.currentDefaultConfig
val originalUncleanLeaderElectionEnable = originalLogConfig.uncleanLeaderElectionEnable
- val newBrokerDefaults = new util.HashMap[String, Object](originalLogConfig.originals)
- newConfig.valuesFromThisConfig.forEach { (k, v) =>
- if (DynamicLogConfig.ReconfigurableConfigs.contains(k)) {
Review Comment:
Good call @rajinisivaram !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org