KAFKA-13224: Ensure that broker.id is set in KafkaConfig#originals #11312
Conversation
Some plugins make use of KafkaConfig#originals rather than the KafkaConfig object. We should ensure that these plugins see the correct value for broker.id if the broker is running in KRaft mode and node.id has been configured, but not broker.id. This PR does this by ensuring that both node.id and broker.id are set in the orignals map if either one is set. We also check that they are set to the same value in KafkaConfig#validateValues. Co-author: Ron Dagostino <rdagostino@confluent.io>
There was a problem hiding this comment.
Thanks @cmccabe (and @rondagostino). Two small things inline
| @@ -1385,10 +1385,22 @@ object KafkaConfig { | |||
| } | |||
| if (maybeSensitive) Password.HIDDEN else value | |||
| } | |||
|
|
|||
| def populateSynonyms(input: util.Map[_, _]): util.Map[Any, Any] = { | |||
There was a problem hiding this comment.
A comment or docstring would be useful here explaining the motivation for this method
| @@ -1385,10 +1385,22 @@ object KafkaConfig { | |||
| } | |||
| if (maybeSensitive) Password.HIDDEN else value | |||
| } | |||
|
|
|||
| def populateSynonyms(input: util.Map[_, _]): util.Map[Any, Any] = { | |||
| val output = new util.HashMap[Any, Any](input) | |||
There was a problem hiding this comment.
Is it safe to make a new map here? Are there any callers of KafkaConfig() which might expect to modify the props map after the object has been constructed?
There was a problem hiding this comment.
AbstractConfig already copies the map which is passed to it.
| } | ||
|
|
||
| class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigOverride: Option[DynamicBrokerConfig]) | ||
| extends AbstractConfig(KafkaConfig.configDef, props, doLog) with Logging { | ||
| extends AbstractConfig(KafkaConfig.configDef, KafkaConfig.populateSynonyms(props), doLog) with Logging { |
There was a problem hiding this comment.
It seems a little bit odd that we populate synonyms only in the reference that we're passing to AbstractConfig. The field KafkaConfig.props could still be accessed directly (maybe it should be private?). Would it make sense to move this to a factory method? For example:
object KafkaConfig {
def apply(props: java.util.Map[_, _], doLog: Boolean, dynamicConfigOverride: Option[DynamicBrokerConfig]): KafkaConfig = {
new KafkaConfig(populateSynonyms(props), doLog, dynamicConfigOverride)
}
}There was a problem hiding this comment.
Yes, good find. This is a case where Scala's behavior is kind of annoying. I wish there was a way to opt-out of the auto-initialization.
Anyway, I made the primary constructor private, and put a call to KafkaConfig#populateSynonyms in all the (publicly visible) secondary constructors. This should fix it...
…pache#11312) Some plugins make use of KafkaConfig#originals rather than the KafkaConfig object. We should ensure that these plugins see the correct value for broker.id if the broker is running in KRaft mode and node.id has been configured, but not broker.id. This PR does this by ensuring that both node.id and broker.id are set in the originals map if either one is set. We also check that they are set to the same value in KafkaConfig#validateValues. Co-author: Ron Dagostino <rdagostino@confluent.io>
Some plugins make use of KafkaConfig#originals rather than the
KafkaConfig object. We should ensure that these plugins see the
correct value for broker.id if the broker is running in KRaft mode and
node.id has been configured, but not broker.id.
This PR does this by ensuring that both node.id and broker.id are set in
the orignals map if either one is set. We also check that they are set
to the same value in KafkaConfig#validateValues.
Co-author: Ron Dagostino rdagostino@confluent.io