KAFKA-18928 Support dynamically changing configs for a specific dynamic quorum controller (#20938)

jiafu1115 · web-flow · commit 45a51fe19e32 · 2026-01-06T02:11:51.000+08:00
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. ``` <img width="1715" height="205" alt="image" src="https://github.com/user-attachments/assets/e0b096b4-2a1c-48dd-8aaf-0cf9855437df" /> ### The result of the fix <img width="1877" height="226" alt="image" src="https://github.com/user-attachments/assets/b09ca3d5-f470-4f43-936c-f8deb209b406" /> Reviewers: Alyssa Huang <ahuang@confluent.io>, TaiJuWu <tjwu1217@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
@@ -32,6 +32,7 @@ type: docs
 
 ### Notable changes in 4.3.0
 
+  * Support dynamically changing configs for dynamic quorum controllers. Previously only brokers and static quorum controllers were supported. For further details, please refer to [KAFKA-18928](https://issues.apache.org/jira/browse/KAFKA-18928). 
   * Two new configs have been introduced: `group.coordinator.cached.buffer.max.bytes` and `share.coordinator.cached.buffer.max.bytes`. They allow the respective coordinators to set the maximum buffer size retained for reuse. For further details, please refer to [KIP-1196](https://cwiki.apache.org/confluence/x/hA5JFg). 
   * The new config have been introduced: `remote.log.metadata.topic.min.isr` with 2 as default value. You can correct the min.insync.replicas for the existed __remote_log_metadata topic via kafka-configs.sh if needed. For further details, please refer to [KIP-1235](https://cwiki.apache.org/confluence/x/yommFw).
 
diff --git a/docs/getting-started/zk2kraft.md b/docs/getting-started/zk2kraft.md
@@ -153,7 +153,7 @@ Note that the entity-type must be specified as `broker-loggers`, even though we
                         
 
   
-It is not currently possible to apply a dynamic configuration on only a single controller. 
+Prior to version 4.3, dynamic configuration updates were not supported unless a static quorum was used. 
 
 
 
diff --git a/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
@@ -339,6 +339,10 @@ Map<Integer, BrokerRegistration> brokerRegistrations() {
         return brokerRegistrations;
     }
 
+    Map<Integer, ControllerRegistration> controllerRegistrations() {
+        return controllerRegistrations;
+    }
+
     /**
      * Process an incoming broker registration request.
      */
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -479,10 +479,8 @@ public void accept(ConfigResource configResource) {
                         throw new InvalidRequestException("Invalid broker name " +
                             configResource.name());
                     }
-                    if (!(clusterControl.brokerRegistrations().containsKey(nodeId) ||
-                            featureControl.isControllerId(nodeId))) {
-                        throw new BrokerIdNotRegisteredException("No node with id " +
-                            nodeId + " found.");
+                    if (!isNodeIdRegistered(nodeId)) {
+                        throw new BrokerIdNotRegisteredException("No node with id " + nodeId + " found.");
                     }
                     break;
                 case TOPIC:
@@ -495,6 +493,19 @@ public void accept(ConfigResource configResource) {
                     break;
             }
         }
+
+        /**
+         * Checks if a node id is registered as a broker, controller in static/dynamic quorum.
+         */
+        private boolean isNodeIdRegistered(int nodeId) {
+            if (clusterControl.brokerRegistrations().containsKey(nodeId)) {
+                return true;
+            }
+            if (featureControl.isControllerId(nodeId)) {
+                return true;
+            }
+            return clusterControl.controllerRegistrations().containsKey(nodeId);
+        }
     }
 
     class QuorumClusterFeatureSupportDescriber implements ClusterFeatureSupportDescriber {
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -1404,6 +1404,32 @@ public void testConfigResourceExistenceChecker() throws Throwable {
         }
     }
 
+    @Test
+    public void testIsNodeIdRegisteredWithDynamicQuorum() throws Throwable {
+        try (
+            MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3).build();
+            QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv).build()
+        ) {
+            QuorumController active = controlEnv.activeController();
+            ConfigResourceExistenceChecker checker = active.new ConfigResourceExistenceChecker();
+            
+            // Register dynamic controller with ID 100
+            active.registerController(ANONYMOUS_CONTEXT,
+                new ControllerRegistrationRequestData()
+                    .setControllerId(100)
+                    .setIncarnationId(Uuid.randomUuid())
+                    .setZkMigrationReady(false)
+                    .setListeners(new ControllerRegistrationRequestData.ListenerCollection())
+                    .setFeatures(new ControllerRegistrationRequestData.FeatureCollection())).get();
+            
+            checker.accept(new ConfigResource(BROKER, "100"));
+            
+            // Unregistered node should throw exception
+            assertThrows(BrokerIdNotRegisteredException.class,
+                () -> checker.accept(new ConfigResource(BROKER, "999")));
+        }
+    }
+
     @Test
     public void testFatalMetadataReplayErrorOnActive() throws Throwable {
         try (
diff --git a/server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java b/server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java
@@ -49,6 +49,8 @@
 import org.apache.kafka.common.errors.MismatchedEndpointTypeException;
 import org.apache.kafka.common.errors.UnsupportedEndpointTypeException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.metrics.KafkaMetric;
+import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.resource.PatternType;
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.resource.ResourceType;
@@ -58,6 +60,7 @@
 import org.apache.kafka.common.test.api.ClusterTestDefaults;
 import org.apache.kafka.common.test.api.Type;
 import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+import org.apache.kafka.network.SocketServerConfigs;
 import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.test.TestUtils;
 
@@ -226,30 +229,91 @@ public void testIncrementalAlterConfigs(ClusterInstance clusterInstance) throws
     }
 
     private void testIncrementalAlterConfigs(ClusterInstance clusterInstance, boolean usingBootstrapControllers) throws Exception {
+        Collection<Integer> nodeIds = usingBootstrapControllers ?
+                clusterInstance.controllerIds() : clusterInstance.brokers().keySet();
         try (Admin admin = Admin.create(adminConfig(clusterInstance, usingBootstrapControllers))) {
-            int nodeId = usingBootstrapControllers ?
-                    clusterInstance.controllers().values().iterator().next().config().nodeId() :
-                    clusterInstance.brokers().values().iterator().next().config().nodeId();
-            ConfigResource nodeResource = new ConfigResource(BROKER, "" + nodeId);
-            ConfigResource defaultResource = new ConfigResource(BROKER, "");
-            Map<ConfigResource, Collection<AlterConfigOp>> alterations = Map.of(
-                    nodeResource, List.of(new AlterConfigOp(new ConfigEntry("my.custom.config", "foo"), AlterConfigOp.OpType.SET)),
-                    defaultResource, List.of(new AlterConfigOp(new ConfigEntry("my.custom.config", "bar"), AlterConfigOp.OpType.SET))
-            );
-            admin.incrementalAlterConfigs(alterations).all().get(1, TimeUnit.MINUTES);
-            TestUtils.retryOnExceptionWithTimeout(30_000, () -> {
-                Config config = admin.describeConfigs(List.of(nodeResource)).
-                        all().get(1, TimeUnit.MINUTES).get(nodeResource);
-                ConfigEntry entry = config.entries().stream().
-                        filter(e -> e.name().equals("my.custom.config")).
-                        findFirst().orElseThrow();
-                assertEquals(DYNAMIC_BROKER_CONFIG, entry.source(),
-                        "Expected entry for my.custom.config to come from DYNAMIC_BROKER_CONFIG. " +
-                                "Instead, the entry was: " + entry);
-            });
+            for (int nodeId : nodeIds) {
+                ConfigResource nodeResource = new ConfigResource(BROKER, "" + nodeId);
+                ConfigResource defaultResource = new ConfigResource(BROKER, "");
+                String nodeMaxConnectionsValue = String.valueOf(1000 + nodeId);
+                String defaultMaxConnectionsValue = String.valueOf(2000 + nodeId);
+                String defaultConnectionRateValue = String.valueOf(2000 + nodeId);
+
+                // Set configs: MAX_CONNECTIONS_CONFIG for per-broker, both configs for default
+                Map<ConfigResource, Collection<AlterConfigOp>> alterations = Map.of(
+                        nodeResource, List.of(
+                                new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, nodeMaxConnectionsValue), AlterConfigOp.OpType.SET)
+                        ),
+                        defaultResource, List.of(
+                                new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, defaultMaxConnectionsValue), AlterConfigOp.OpType.SET),
+                                new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTION_CREATION_RATE_CONFIG, defaultConnectionRateValue), AlterConfigOp.OpType.SET)
+                        )
+                );
+                admin.incrementalAlterConfigs(alterations).all().get(1, TimeUnit.MINUTES);
+                
+                // Verify per-broker configs: MAX_CONNECTIONS_CONFIG and MAX_CONNECTION_CREATION_RATE_CONFIG
+                verifyConfigValue(admin, nodeResource, SocketServerConfigs.MAX_CONNECTIONS_CONFIG,
+                        DYNAMIC_BROKER_CONFIG, nodeMaxConnectionsValue);
+                verifyConfigValue(admin, nodeResource, SocketServerConfigs.MAX_CONNECTION_CREATION_RATE_CONFIG,
+                        DYNAMIC_DEFAULT_BROKER_CONFIG, defaultConnectionRateValue);
+                
+                // Verify default broker configs: MAX_CONNECTIONS_CONFIG and MAX_CONNECTION_CREATION_RATE_CONFIG
+                verifyConfigValue(admin, defaultResource, SocketServerConfigs.MAX_CONNECTIONS_CONFIG,
+                        DYNAMIC_DEFAULT_BROKER_CONFIG, defaultMaxConnectionsValue);
+                verifyConfigValue(admin, defaultResource, SocketServerConfigs.MAX_CONNECTION_CREATION_RATE_CONFIG,
+                        DYNAMIC_DEFAULT_BROKER_CONFIG, defaultConnectionRateValue);
+
+                Object node = (usingBootstrapControllers ? clusterInstance.controllers() : clusterInstance.brokers()).get(nodeId);
+                // Verify that SocketServer has actually updated the max connection limit in ConnectionQuotas
+                // The per-broker config should take precedence over the default config
+                verifySocketServerMaxConnectionsUpdated(node, Integer.parseInt(nodeMaxConnectionsValue));
+                // Verify MAX_CONNECTION_CREATION_RATE_CONFIG is also updated
+                // The default config value should be used (per-broker config doesn't set this)
+                verifySocketServerMaxConnectionCreationRateUpdated(node, Integer.parseInt(defaultConnectionRateValue));
+            }
         }
     }
 
+    private void verifySocketServerMaxConnectionsUpdated(Object node, int expectedMaxConnections) throws Exception {
+        Object socketServer = node.getClass().getMethod("socketServer").invoke(node);
+        Object connectionQuotas = socketServer.getClass().getMethod("connectionQuotas").invoke(socketServer);
+        java.lang.reflect.Field field = connectionQuotas.getClass().getDeclaredField("brokerMaxConnections");
+        field.setAccessible(true);
+        int actualMaxConnections = ((Number) field.get(connectionQuotas)).intValue();
+        assertEquals(expectedMaxConnections, actualMaxConnections,
+                "SocketServer ConnectionQuotas.brokerMaxConnections should be " + expectedMaxConnections +
+                " but was " + actualMaxConnections);
+    }
+    
+    private void verifySocketServerMaxConnectionCreationRateUpdated(Object node, int expectedMaxConnectionCreationRate) throws Exception {
+        Metrics metrics = (Metrics) node.getClass().getMethod("metrics").invoke(node);
+        KafkaMetric metric = metrics.metrics().entrySet().stream()
+                .filter(entry -> "broker-connection-accept-rate".equals(entry.getKey().name()))
+                .map(java.util.Map.Entry::getValue)
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("Broker connection rate metric not found"));
+        double actualBound = metric.config().quota().bound();
+        assertEquals(expectedMaxConnectionCreationRate, actualBound,
+                "Connection creation rate quota should be " + expectedMaxConnectionCreationRate + " but was " + actualBound);
+    }
+    
+    private void verifyConfigValue(Admin admin, ConfigResource resource, String configName,
+                                   org.apache.kafka.clients.admin.ConfigEntry.ConfigSource expectedSource,
+                                   String expectedValue) throws Exception {
+        TestUtils.retryOnExceptionWithTimeout(30_000, () -> {
+            Config config = admin.describeConfigs(List.of(resource)).all().get(1, TimeUnit.MINUTES).get(resource);
+            ConfigEntry entry = config.entries().stream()
+                    .filter(e -> e.name().equals(configName))
+                    .findFirst().orElseThrow();
+            assertEquals(expectedSource, entry.source(),
+                    "Expected entry for " + configName + " to come from " + expectedSource +
+                    ". Instead, the entry was: " + entry);
+            assertEquals(expectedValue, entry.value(),
+                    "Expected value for " + configName + " to be " + expectedValue +
+                    ". Instead, the value was: " + entry.value());
+        });
+    }
+
     @ClusterTest(brokers = 3)
     public void testAlterReassignmentsWithBootstrapControllers(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
         String topicName = "foo";
@@ -357,4 +421,14 @@ public void testDescribeConfigs(ClusterInstance clusterInstance) throws Exceptio
             assertEquals("2", configEntry.value());
         }
     }
+
+    @ClusterTest(controllers = 1, standalone = true)
+    public void testIncrementalAlterConfigsBySingleControllerWithDynamicQuorum(ClusterInstance clusterInstance) throws Exception {
+        testIncrementalAlterConfigs(clusterInstance, true);
+    }
+
+    @ClusterTest(controllers = 3, standalone = true)
+    public void testIncrementalAlterConfigsByAllControllersWithDynamicQuorum(ClusterInstance clusterInstance) throws Exception {
+        testIncrementalAlterConfigs(clusterInstance, true);
+    }
 }
diff --git a/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterConfig.java b/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterConfig.java
@@ -59,13 +59,15 @@ public class ClusterConfig {
     private final List<String> tags;
     private final Map<Integer, Map<String, String>> perServerProperties;
     private final Map<Feature, Short> features;
+    private final boolean standalone;
 
     @SuppressWarnings("checkstyle:ParameterNumber")
     private ClusterConfig(Set<Type> types, int brokers, int controllers, int disksPerBroker, boolean autoStart,
                   SecurityProtocol brokerSecurityProtocol, ListenerName brokerListenerName,
                   SecurityProtocol controllerSecurityProtocol, ListenerName controllerListenerName, File trustStoreFile,
                   MetadataVersion metadataVersion, Map<String, String> serverProperties,
-                  Map<Integer, Map<String, String>> perServerProperties, List<String> tags, Map<Feature, Short> features) {
+                  Map<Integer, Map<String, String>> perServerProperties, List<String> tags, Map<Feature, Short> features,
+                  boolean standalone) {
         // do fail fast. the following values are invalid for kraft modes.
         if (brokers < 0) throw new IllegalArgumentException("Number of brokers must be greater or equal to zero.");
         if (controllers < 0) throw new IllegalArgumentException("Number of controller must be greater or equal to zero.");
@@ -86,6 +88,7 @@ private ClusterConfig(Set<Type> types, int brokers, int controllers, int disksPe
         this.perServerProperties = Objects.requireNonNull(perServerProperties);
         this.tags = Objects.requireNonNull(tags);
         this.features = Objects.requireNonNull(features);
+        this.standalone = standalone;
     }
 
     public Set<Type> clusterTypes() {
@@ -136,6 +139,10 @@ public MetadataVersion metadataVersion() {
         return metadataVersion;
     }
 
+    public boolean standalone() {
+        return standalone;
+    }
+
     public Map<Integer, Map<String, String>> perServerOverrideProperties() {
         return perServerProperties;
     }
@@ -192,7 +199,8 @@ public static Builder builder(ClusterConfig clusterConfig) {
                 .setServerProperties(clusterConfig.serverProperties)
                 .setPerServerProperties(clusterConfig.perServerProperties)
                 .setTags(clusterConfig.tags)
-                .setFeatures(clusterConfig.features);
+                .setFeatures(clusterConfig.features)
+                .setStandalone(clusterConfig.standalone);
     }
 
     public static class Builder {
@@ -211,6 +219,7 @@ public static class Builder {
         private Map<Integer, Map<String, String>> perServerProperties = Map.of();
         private List<String> tags = List.of();
         private Map<Feature, Short> features = Map.of();
+        private boolean standalone = false;
 
         private Builder() {}
 
@@ -291,10 +300,15 @@ public Builder setFeatures(Map<Feature, Short> features) {
             return this;
         }
 
+        public Builder setStandalone(boolean standalone) {
+            this.standalone = standalone;
+            return this;
+        }
+
         public ClusterConfig build() {
             return new ClusterConfig(types, brokers, controllers, disksPerBroker, autoStart,
                     brokerSecurityProtocol, brokerListenerName, controllerSecurityProtocol, controllerListenerName,
-                    trustStoreFile, metadataVersion, serverProperties, perServerProperties, tags, features);
+                    trustStoreFile, metadataVersion, serverProperties, perServerProperties, tags, features, standalone);
         }
     }
 }
diff --git a/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java b/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java
@@ -57,4 +57,5 @@
     // users can add tags that they want to display in test
     String[] tags() default {};
     ClusterFeature[] features() default {};
+    boolean standalone() default false;
 }
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/ClusterTestExtensions.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/ClusterTestExtensions.java
@@ -294,6 +294,7 @@ private List<TestTemplateInvocationContext> processClusterTestInternal(
             .setMetadataVersion(clusterTest.metadataVersion())
             .setTags(List.of(clusterTest.tags()))
             .setFeatures(features)
+            .setStandalone(clusterTest.standalone())
             .build();
 
         return Arrays.stream(types)
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
@@ -324,6 +324,7 @@ public void format() throws Exception {
                 KafkaClusterTestKit.Builder builder = new KafkaClusterTestKit.Builder(nodes);
                 // Copy properties into the TestKit builder
                 clusterConfig.serverProperties().forEach(builder::setConfigProp);
+                builder.setStandalone(clusterConfig.standalone());
                 this.clusterTestKit = builder.build();
                 this.clusterTestKit.format();
             }

Original file line number	Diff line number	Diff line change
@@ -153,7 +153,7 @@ Note that the entity-type must be specified as `broker-loggers`, even though we
`153`	`153`
`154`	`154`
`155`	`155`
`156`		`-It is not currently possible to apply a dynamic configuration on only a single controller.`
	`156`	`+Prior to version 4.3, dynamic configuration updates were not supported unless a static quorum was used.`
`157`	`157`
`158`	`158`
`159`	`159`
Original file line number	Diff line number	Diff line change
`@@ -57,4 +57,5 @@`
`57`	`57`	`// users can add tags that they want to display in test`
`58`	`58`	`String[] tags() default {};`
`59`	`59`	`ClusterFeature[] features() default {};`
	`60`	`+ boolean standalone() default false;`
`60`	`61`	`}`
Original file line number	Diff line number	Diff line change
`@@ -324,6 +324,7 @@ public void format() throws Exception {`
`324`	`324`	`KafkaClusterTestKit.Builder builder = new KafkaClusterTestKit.Builder(nodes);`
`325`	`325`	`// Copy properties into the TestKit builder`
`326`	`326`	`clusterConfig.serverProperties().forEach(builder::setConfigProp);`
	`327`	`+ builder.setStandalone(clusterConfig.standalone());`
`327`	`328`	`this.clusterTestKit = builder.build();`
`328`	`329`	`this.clusterTestKit.format();`
`329`	`330`	`}`