KAFKA-18928: Support dynamically changing configs for a specific dynamic quorum controller

[jiafu1115]

Fix https://issues.apache.org/jira/browse/KAFKA-18928

Currently. It only support dynamically changing configs for a specific static quorum controller.

Test Environment

Dynamic quorum environment: controller.quorum.bootstrap.servers=ip1:port,ip2:port,ip3:port Static quorum environment: controller.quorum.voters=id1@ip1:port,id2@ip2:port,id3@ip3:port

Test command

./kafka-configs.sh--bootstrap-controller ip:port --entity-type brokers --entity-name 1001 --command-config /etc/kafka/admin_config.properties --alter --add-config max.connections=10086

Completed updating config for broker 1001.

Verify command

./kafka-configs.sh --bootstrap-controller ip:port --entity-type brokers --entity-name 1001 --command-config /etc/kafka/admin_config.properties --describe --all |grep -E 'max.connections|node'

max.connections=10086 sensitive=false synonyms={DYNAMIC_BROKER_CONFIG:max.connections=100005, DEFAULT_CONFIG:max.connections=2147483647} node.id=1001 sensitive=false synonyms={STATIC_BROKER_CONFIG:node.id=1001}

Test result

The issue on the dynamic quorum:

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.BrokerIdNotRegisteredException: No node with id 1001 found.
        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:173)
        at kafka.admin.ConfigCommand$.alterResourceConfig(ConfigCommand.scala:425)
        at kafka.admin.ConfigCommand$.alterConfig(ConfigCommand.scala:192)
        at kafka.admin.ConfigCommand$.processCommand(ConfigCommand.scala:165)
        at kafka.admin.ConfigCommand$.main(ConfigCommand.scala:87)
        at kafka.admin.ConfigCommand.main(ConfigCommand.scala)
Caused by: org.apache.kafka.common.errors.BrokerIdNotRegisteredException: No node with id 1001 found.

The result of the fix

[github-actions[bot]]

A label of 'needs-attention' was automatically added to this PR in order to raise the attention of the committers. Once this issue has been triaged, the triage label should be removed to prevent this automation from happening again.

[jiafu1115]

Test: Dynamic Controller: Only change one and verify if others also been changed.

[2025-12-05 23:52:39,776] INFO [SocketServer listenerType=CONTROLLER, nodeId=1001] Updating broker-wide maxConnections: 123456789 (kafka.network.SocketServer)

For static controller. The test result is similar . It will be workable without this code change. (I also thought it should not work according to the document ).

@ahuang98 Thanks for your review. I attach this test according to your suggestion. Help to check again. Thanks

[ahuang98]

@jiafu1115 thanks for the response and for testing it out manually!

Good to see that the config isn't picked up by the other controllers. I wanted to clarify that my bigger concern (maybe uninformed concern since I'm not very familiar with config code) is that we are able to set the config for individual controllers, but they might not actually take effect. For instance, maybe kraft correctly creates a config record in the metadata log that broker id 1001 has this new config, but is controller 1001 actually able to apply this config (would it actually limit max connections at the new value)?

And, we probably need additional tests in the code for the dynamic config update (for instance, check out DynamicConfigChangeTest)

[jiafu1115]

@ahuang98 thanks for reviewing again. you can check this code lines and compare with the log I pasted in last comment (I copied it here) [2025-12-05 23:52:39,776] INFO [SocketServer listenerType=CONTROLLER, nodeId=1001] Updating broker-wide maxConnections: 123456789 (kafka.network.SocketServer)

For the unit test. I can take some time to check if we can add the extra one for it. BTW: the root cause why I don't add much more test cases at the beginning is that the change is very small which allow more capacity. It don't change any existed logic. But I test it manually is due to I need to make sure it can work end to end.

thanks

[jiafu1115]

org.apache.kafka.server.BootstrapControllersIntegrationTest#testIncrementalAlterConfigsByControllers the existed test to prove static quorum can work

[chia7712]

And, we probably need additional tests in the code for the dynamic config update (for instance, check out DynamicConfigChangeTest)

I agree. We should create a helper test to validate all dynamic configurations on the controller. We must also cover both static quorum and dynamic quorum. ClusterTest supports static quorum. The dynamic quorum tests could be implemented by KafkaClusterTestKit

[jiafu1115]

@chia7712 @ahuang98 add the related core test. thanks ! FYI: Before the code change:

[chia7712]

@jiafu1115 could you please fix the conflicts?

[jiafu1115]

@jiafu1115 could you please fix the conflicts?

@chia7712 done. thanks

[apalan60]

Thanks for the PR. I tested it locally with a dynamic voter setup, and it successfully updates a single controller’s config without any exceptions.

./bin/kafka-configs.sh --bootstrap-controller localhost:9093 --entity-type brokers --entity-name 2 --describe --all | grep log.cleaner.threads
./bin/kafka-configs.sh --bootstrap-controller localhost:9093 --entity-type brokers --entity-name 2 --alter --add-config log.cleaner.threads=6
./bin/kafka-configs.sh --bootstrap-controller localhost:9093 --entity-type brokers --entity-name 2 --describe --all | grep log.cleaner.threads

log.cleaner.threads=1 sensitive=false synonyms={DEFAULT_CONFIG:log.cleaner.threads=1}
Completed updating config for broker 2.
log.cleaner.threads=6 sensitive=false synonyms={DYNAMIC_BROKER_CONFIG:log.cleaner.threads=6, DEFAULT_CONFIG:log.cleaner.threads=1}

[jiafu1115]

@apalan60 Thank you!

[jiafu1115]

Attach additional test results based on @chia7712 's proposed doc's change: (not directly related to code change; recorded here to avoid forgetting this logic.)

There is no any method to change all controllers in only one time. change with --entity-default only change the default value. it won't change the final value if the value had been set.

1 alter command should contain entity-name or --entity-default in alter command root@ip-10-20-1-107:/usr/local/kafka/bin# ./kafka-configs.sh --bootstrap-controller --bootstrap-controller 10.20.1.107:9889 --entity-type brokers --command-config /etc/kafka/admin_config.properties --alter --add-config max.connections=10086 An entity-name or default entity must be specified with --alter of users, clients, brokers or ips

2 --entity-default is to update the default value: DYNAMIC_DEFAULT_BROKER_CONFIG instead of the all controller. root@ip-10-20-1-107:/usr/local/kafka/bin# ./kafka-configs.sh --bootstrap-controller 10.20.1.107:9889 --entity-type brokers --entity-default --command-config /etc/kafka/admin_config.properties --alter --add-config max.connections=10086 Completed updating default config for brokers in the cluster. root@ip-10-20-1-107:/usr/local/kafka/bin# ./kafka-configs.sh --bootstrap-controller 10.20.1.107:9889 --entity-type brokers --entity-name 1001 --command-config /etc/kafka/admin_config.properties --describe --all |grep -E 'max.connections|node' max.connections=10086 sensitive=false synonyms={DYNAMIC_DEFAULT_BROKER_CONFIG:max.connections=10086, DEFAULT_CONFIG:max.connections=2147483647}

3 alert with --entity-name will override the default value. (note: after changed. if alter with --entity-default. it won't change the final value. it only change the default value) root@ip-10-20-1-107:/usr/local/kafka/bin# ./kafka-configs.sh --bootstrap-controller 10.20.1.107:9889 --entity-type brokers --entity-name 1001 --command-config /etc/kafka/admin_config.properties --alter --add-config max.connections=10000 Completed updating config for broker 1001. root@ip-10-20-1-107:/usr/local/kafka/bin# ./kafka-configs.sh --bootstrap-controller 10.20.1.107:9889 --entity-type brokers --entity-name 1001 --command-config /etc/kafka/admin_config.properties --describe --all |grep -E 'max.connections|node' max.connections=10000 sensitive=false synonyms={DYNAMIC_BROKER_CONFIG:max.connections=10000, DYNAMIC_DEFAULT_BROKER_CONFIG:max.connections=10086, DEFAULT_CONFIG:max.connections=2147483647}

[chia7712]

@jiafu1115 please fix the docs

[jiafu1115]

@jiafu1115 please fix the docs

@chia7712 done. thanks