KAFKA-13417; Ensure dynamic reconfigurations set old config properly#11448
KAFKA-13417; Ensure dynamic reconfigurations set old config properly#11448hachikuji merged 6 commits intoapache:trunkfrom
Conversation
|
Note I am planning to add some additional test cases here. We should cover all of the |
dajac
left a comment
dajac
left a comment
There was a problem hiding this comment.
Good catch! Left a small comment.
|
|
||
| // We need a copy of the current config since `currentConfig` is initialized with `kafkaConfig` | ||
| // which means the call to `updateCurrentConfig` would end up mutating `oldConfig`. | ||
| val oldConfig = if (kafkaConfig eq currentConfig) { |
There was a problem hiding this comment.
@hachikuji Thanks for the PR. Do you know if there is an issue only with KRraft or is there an issue with ZK as well?
With ZK, the sequence is:
- Create KafkaConfig with static configs from server.properties
- Initialize KafkaConfig with initial configs from ZooKeeper using
DynamicBrokerConfig.initialize(zkClient). This always creates a new KafkaConfig, so we don't need this check? - Start DynamicConfigManager. All ZK updates after 2) are handled through change notifications.
With KRaft, there doesn't seem to be an initialize() for initializing the state from existing dynamic configs, so are all configs handled similar to 3)? In which case, we should perhaps move the initialization of currentConfig from DynamicBrokerConfig.initialize(zkClient) to somewhere common for KRaft and avoid this check here?
There was a problem hiding this comment.
@rajinisivaram Hmm, good point. It looks like the bug only affects KRaft since there is no call to initialize. The code still feels a little slippery though, so maybe there is room for some defensiveness. Let me take a look.
There was a problem hiding this comment.
I pushed a patch which initializes currentConfig as null in order to make the call to initialize required. Let me know if that seems like a reasonable approach.
rajinisivaram
left a comment
rajinisivaram
left a comment
There was a problem hiding this comment.
@hachikuji Thanks for the updates, LGTM.
| private val dynamicConfigPasswordEncoder = maybeCreatePasswordEncoder(kafkaConfig.passwordEncoderSecret) | ||
|
|
||
| private[server] def initialize(zkClient: KafkaZkClient): Unit = { | ||
| private[server] def initialize(zkClientOpt: Option[KafkaZkClient]): Unit = { |
There was a problem hiding this comment.
In the non-ZK world, does the initialization using existing dynamic configs happen later? Or does the original Kafka config contain the existing dynamic configs as well?
There was a problem hiding this comment.
It looks like dynamic configs get initialized later when we load the metadata log. I think this probably could be improved. Do you see any specific issues?
There was a problem hiding this comment.
In the ZK world, all ZK configs are static configs. We initialize dynamic configs very early on, once we have a ZK client that we create with ZK configs.
For the non-ZK world, I wasn't sure if the initialization required some dynamic configs. For example, we allow SSL keystore passwords to be stored in ZK prior to starting up brokers to avoid specifying any passwords in server.properties. Do we ensure that SocketServer only starts up after dynamic configs are initialized with KRaft?
There was a problem hiding this comment.
Colin has an incoming KIP which will cover secret storage. It is a known gap at the moment.
rajinisivaram
left a comment
rajinisivaram
left a comment
There was a problem hiding this comment.
@hachikuji Thanks for the updates, LGTM
…ntegration-11-nov * ak/trunk: (15 commits) KAFKA-13429: ignore bin on new modules (apache#11415) KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides (apache#11272) KAFKA-12487: Add support for cooperative consumer protocol with sink connectors (apache#10563) MINOR: Log client disconnect events at INFO level (apache#11449) MINOR: Remove topic null check from `TopicIdPartition` and adjust constructor order (apache#11403) KAFKA-13417; Ensure dynamic reconfigurations set old config properly (apache#11448) MINOR: Adding a constant to denote UNKNOWN leader in LeaderAndEpoch (apache#11477) KAFKA-10543: Convert KTable joins to new PAPI (apache#11412) KAFKA-12226: Commit source task offsets without blocking on batch delivery (apache#11323) KAFKA-13396: Allow create topic without partition/replicaFactor (apache#11429) ...
…11448) This patch fixes a bug in `DynamicBrokerConfig` which causes some configuration changes to be ignored. In particular, the bug is the result of the reference to the old configuration getting indirectly mutated prior to the call to `BrokerReconfigurable.reconfigure`. This causes the first dynamic configuration update to pass effectively the same configuration as both `oldConfig` and `newConfig`. In cases such as in `DynamicThreadPool`, the update is ignored because the old configuration value matches the new configuration value. This bug only affects KRaft. It is protected in the zk broker by the call to `DynamicBrokerConfig.initialize()`, which overwrites the stored reference to the original configuration. The patch fixes the problem by ensuring that `initialize()` is also invoked in KRaft when `BrokerServer` starts up. Reviewers: David Jacot <djacot@confluent.io>, Rajini Sivaram <rajinisivaram@googlemail.com>
…pache#11448) This patch fixes a bug in `DynamicBrokerConfig` which causes some configuration changes to be ignored. In particular, the bug is the result of the reference to the old configuration getting indirectly mutated prior to the call to `BrokerReconfigurable.reconfigure`. This causes the first dynamic configuration update to pass effectively the same configuration as both `oldConfig` and `newConfig`. In cases such as in `DynamicThreadPool`, the update is ignored because the old configuration value matches the new configuration value. This bug only affects KRaft. It is protected in the zk broker by the call to `DynamicBrokerConfig.initialize()`, which overwrites the stored reference to the original configuration. The patch fixes the problem by ensuring that `initialize()` is also invoked in KRaft when `BrokerServer` starts up. Reviewers: David Jacot <djacot@confluent.io>, Rajini Sivaram <rajinisivaram@googlemail.com>
3 participants
This patch fixes a bug in
DynamicBrokerConfigwhich causes some configuration changes to be ignored. In particular, the bug is the result of the reference to the old configuration getting indirectly mutated prior to the call toBrokerReconfigurable.reconfigure. This causes the first dynamic configuration update to pass effectively the same configuration as botholdConfigandnewConfig. In cases such as inDynamicThreadPool, the update is ignored because the old configuration value matches the new configuration value.Committer Checklist (excluded from commit message)