Allow per listener configurations (#12239)

tinaselenge · web-flow · commit 2d31a2b6e7d7 · 2026-01-26T21:08:17.000+01:00
Signed-off-by: Gantigmaa Selenge &lt;tina.selenge@gmail.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,7 +2,7 @@
 
 ## 0.51.0
 
-* Nothing here yet, but we will surely develop something new pretty soon 😉
+* Allow setting the following configurations with the listener prefix (e.g. `listener.name.listener1-9900.`): `connections.max.reauth.ms`, `max.connections*` and `max.connection.creation.rate`.
 
 ### Major changes, deprecations, and removals
 
diff --git a/api/src/main/java/io/strimzi/api/kafka/model/kafka/KafkaClusterSpec.java b/api/src/main/java/io/strimzi/api/kafka/model/kafka/KafkaClusterSpec.java
@@ -58,6 +58,7 @@
 @EqualsAndHashCode
 @ToString
 public class KafkaClusterSpec implements HasConfigurableMetrics, HasConfigurableLogging, HasJmxOptions, HasReadinessProbe, HasLivenessProbe, UnknownPropertyPreserving {
+    /* When FORBIDDEN_PREFIXES is updated, documentation/api/io.strimzi.api.kafka.model.kafka.KafkaClusterSpec.adoc must be updated accordingly. */
     public static final String FORBIDDEN_PREFIXES = "listeners, advertised., broker., listener., host.name, port, "
             + "inter.broker.listener.name, sasl., ssl., security., password., log.dir, "
             + "zookeeper.connect, zookeeper.set.acl, zookeeper.ssl, zookeeper.clientCnxnSocket, authorizer., super.user, "
@@ -67,6 +68,7 @@ public class KafkaClusterSpec implements HasConfigurableMetrics, HasConfigurable
             + "client.quota.callback.static.storage.per.volume.limit.min.available., client.quota.callback.static.excluded.principal.name.list, "
             + "prometheus.metrics.reporter.";
 
+    /* When FORBIDDEN_PREFIX_EXCEPTIONS is updated, documentation/api/io.strimzi.api.kafka.model.kafka.KafkaClusterSpec.adoc must be updated accordingly. */
     public static final String FORBIDDEN_PREFIX_EXCEPTIONS = "zookeeper.connection.timeout.ms, sasl.server.max.receive.size, "
             + "ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols, ssl.secure.random.implementation, "
             + "cruise.control.metrics.topic.num.partitions, cruise.control.metrics.topic.replication.factor, cruise.control.metrics.topic.retention.ms, "
@@ -75,6 +77,9 @@ public class KafkaClusterSpec implements HasConfigurableMetrics, HasConfigurable
             + "broker.session.timeout.ms, broker.heartbeat.interval.ms, controller.socket.timeout.ms, "
             + "controller.quorum.election.backoff.max.ms, controller.quorum.election.timeout.ms, controller.quorum.fetch.timeout.ms"; // KRaft options
 
+    /* When ALLOWED_PER_LISTENER_CONFIGS is updated, documentation/api/io.strimzi.api.kafka.model.kafka.KafkaClusterSpec.adoc must be updated accordingly. */
+    public static final String ALLOWED_PER_LISTENER_CONFIGS  = "connections.max.reauth.ms, max.connections, max.connection.creation.rate";
+
     private Storage storage;
     private String version;
     private String metadataVersion;
@@ -119,7 +124,7 @@ public void setMetadataVersion(String metadataVersion) {
         this.metadataVersion = metadataVersion;
     }
 
-    @Description("Kafka broker config properties with the following prefixes cannot be set: " + FORBIDDEN_PREFIXES + " (with the exception of: " + FORBIDDEN_PREFIX_EXCEPTIONS + ").")
+    @Description("Kafka broker config properties with certain prefixes cannot be set unless it is in the exception list. Consult the documentation for the list of forbidden prefixes and exceptions")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     public Map<String, Object> getConfig() {
         return config;
diff --git a/api/src/test/resources/crds/v1/040-Crd-kafka.yaml b/api/src/test/resources/crds/v1/040-Crd-kafka.yaml
@@ -379,7 +379,7 @@ spec:
                   config:
                     x-kubernetes-preserve-unknown-fields: true
                     type: object
-                    description: "Kafka broker config properties with the following prefixes cannot be set: listeners, advertised., broker., listener., host.name, port, inter.broker.listener.name, sasl., ssl., security., password., log.dir, zookeeper.connect, zookeeper.set.acl, zookeeper.ssl, zookeeper.clientCnxnSocket, authorizer., super.user, cruise.control.metrics.topic, cruise.control.metrics.reporter.bootstrap.servers, node.id, process.roles, controller., metadata.log.dir, zookeeper.metadata.migration.enable, client.quota.callback.static.kafka.admin., client.quota.callback.static.produce, client.quota.callback.static.fetch, client.quota.callback.static.storage.per.volume.limit.min.available., client.quota.callback.static.excluded.principal.name.list, prometheus.metrics.reporter. (with the exception of: zookeeper.connection.timeout.ms, sasl.server.max.receive.size, ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols, ssl.secure.random.implementation, cruise.control.metrics.topic.num.partitions, cruise.control.metrics.topic.replication.factor, cruise.control.metrics.topic.retention.ms, cruise.control.metrics.topic.auto.create.retries, cruise.control.metrics.topic.auto.create.timeout.ms, cruise.control.metrics.topic.min.insync.replicas, broker.session.timeout.ms, broker.heartbeat.interval.ms, controller.socket.timeout.ms, controller.quorum.election.backoff.max.ms, controller.quorum.election.timeout.ms, controller.quorum.fetch.timeout.ms)."
+                    description: Kafka broker config properties with certain prefixes cannot be set unless it is in the exception list. Consult the documentation for the list of forbidden prefixes and exceptions.
                   authorization:
                     type: object
                     properties:
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaCluster.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaCluster.java
@@ -345,9 +345,10 @@ public static KafkaCluster fromCrd(Reconciliation reconciliation,
                 result.ownerReference,
                 kafkaClusterSpec
         );
+        List<GenericKafkaListener> listeners = kafkaClusterSpec.getListeners();
 
         // Handle Kafka broker configuration
-        KafkaConfiguration configuration = new KafkaConfiguration(reconciliation, kafkaClusterSpec.getConfig().entrySet());
+        KafkaConfiguration configuration = new KafkaConfiguration(reconciliation, kafkaClusterSpec.getConfig().entrySet(), listeners);
         validateConfiguration(reconciliation, kafka, result.kafkaVersion, configuration);
 
         if (kafkaClusterSpec.getQuotas() != null) {
@@ -364,7 +365,6 @@ public static KafkaCluster fromCrd(Reconciliation reconciliation,
             LOGGER.errorCr(reconciliation, "The required field .spec.kafka.listeners is missing");
             throw new InvalidResourceException("The required field .spec.kafka.listeners is missing");
         }
-        List<GenericKafkaListener> listeners = kafkaClusterSpec.getListeners();
         ListenersValidator.validate(reconciliation, result.brokerNodes(), listeners);
         result.listeners = listeners;
 
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaConfiguration.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaConfiguration.java
@@ -6,6 +6,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.strimzi.api.kafka.model.kafka.KafkaClusterSpec;
+import io.strimzi.api.kafka.model.kafka.listener.GenericKafkaListener;
 import io.strimzi.kafka.config.model.ConfigModel;
 import io.strimzi.kafka.config.model.ConfigModels;
 import io.strimzi.operator.common.Reconciliation;
@@ -30,13 +31,16 @@ public class KafkaConfiguration extends AbstractConfiguration {
      */
     public static final String DEFAULT_REPLICATION_FACTOR = "default.replication.factor";
 
+    private static final String PER_LISTENER_CONFIG_FORMAT = "listener.name.%s.%s";
     private static final List<String> FORBIDDEN_PREFIXES;
     private static final List<String> FORBIDDEN_PREFIX_EXCEPTIONS;
+    private static final List<String> ALLOWED_PER_LISTENER_CONFIGURATIONS;
     private static final Map<String, String> DEFAULTS;
 
     static {
         FORBIDDEN_PREFIXES = AbstractConfiguration.splitPrefixesOrOptionsToList(KafkaClusterSpec.FORBIDDEN_PREFIXES);
         FORBIDDEN_PREFIX_EXCEPTIONS = AbstractConfiguration.splitPrefixesOrOptionsToList(KafkaClusterSpec.FORBIDDEN_PREFIX_EXCEPTIONS);
+        ALLOWED_PER_LISTENER_CONFIGURATIONS = AbstractConfiguration.splitPrefixesOrOptionsToList(KafkaClusterSpec.ALLOWED_PER_LISTENER_CONFIGS);
 
         DEFAULTS = new HashMap<>(1);
         // when users remove "min.insync.replicas" from the Kafka custom resource, the operator is going to force the
@@ -65,11 +69,31 @@ public KafkaConfiguration(Reconciliation reconciliation, Iterable<Map.Entry<Stri
         super(reconciliation, jsonOptions, FORBIDDEN_PREFIXES, FORBIDDEN_PREFIX_EXCEPTIONS, List.of(), DEFAULTS);
     }
 
+    /**
+     * Constructor used to instantiate this class from JsonObject. Should be used to create configuration from
+     * ConfigMap / CRD. It also modifies the FORBIDDEN_PREFIX_EXCEPTIONS with the given listener names.
+     *
+     * @param reconciliation        The reconciliation
+     * @param jsonOptions           Json object with configuration options as key ad value pairs.
+     * @param listeners             Listener names to add to the exception list
+     */
+    public KafkaConfiguration(Reconciliation reconciliation, Iterable<Map.Entry<String, Object>> jsonOptions, List<GenericKafkaListener> listeners) {
+        super(reconciliation, jsonOptions, FORBIDDEN_PREFIXES, modifyWithListeners(listeners), List.of(), DEFAULTS);
+    }
+
     private KafkaConfiguration(Reconciliation reconciliation, String configuration, List<String> forbiddenPrefixes) {
         super(reconciliation, configuration, forbiddenPrefixes, List.of(), List.of(), DEFAULTS);
     }
 
 
+    /* test */ static List<String> modifyWithListeners(List<GenericKafkaListener> listeners) {
+        List<String> exceptions = new ArrayList<>();
+        listeners.stream().map(ListenersUtils::identifier).forEach(l -> ALLOWED_PER_LISTENER_CONFIGURATIONS.forEach(c -> exceptions.add(String.format(PER_LISTENER_CONFIG_FORMAT, l, c))));
+        exceptions.addAll(FORBIDDEN_PREFIX_EXCEPTIONS);
+        return exceptions;
+    }
+
+
     /**
      * Returns a KafkaConfiguration created without forbidden option filtering.
      *
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceAssemblyOperator.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceAssemblyOperator.java
@@ -918,7 +918,6 @@ private void handleUserTaskStatusResponse(Reconciliation reconciliation, CruiseC
                     if (taskStatusJson.has(CruiseControlRebalanceKeys.LOAD_BEFORE_OPTIMIZATION.getKey()) &&
                             taskStatusJson.has(CruiseControlRebalanceKeys.LOAD_AFTER_OPTIMIZATION.getKey())) {
                         LOGGER.infoCr(reconciliation, "Rebalance ({}) optimization proposal is now ready", sessionId);
-                        System.out.println("Status" + taskStatusJson);
                         p.complete(buildRebalanceStatus(kafkaRebalance, sessionId, KafkaRebalanceState.ProposalReady, taskStatusJson, conditions));
                         break;
                     }
diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaClusterTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaClusterTest.java
@@ -1891,6 +1891,45 @@ public void testReplicasAndRelatedOptionsValidationOk() {
         });
     }
 
+    @Test
+    public void testPerListenerConfigurations() {
+        Kafka kafkaAssembly = new KafkaBuilder(KAFKA)
+                .editSpec()
+                .editKafka()
+                .withConfig(Map.of("listener.name.tls-9093.max.connections", 1000,
+                        "listener.name.tls-9093.connections.max.reauth.ms", 1000,
+                        "listener.name.tls-9093.connections.max.idle.ms", 100000,
+                        "listener.name.plain-9092.max.connections", 1000,
+                        "listener.name.plain-9092.connections.max.reauth.ms", 1000,
+                        "listener.name.plain-9092.connections.max.idle.ms", 100000,
+                        "listener.name.CONTROLPLANE-9090.max.connections", 1000,
+                        "listener.name.REPLICATION-9091.connections.max.reauth.ms", 1000,
+                        "listener.name.does-not-exist.connections.max.reauth.ms", 1000))
+                .endKafka()
+                .endSpec()
+                .build();
+
+        List<KafkaPool> pools = NodePoolUtils.createKafkaPools(Reconciliation.DUMMY_RECONCILIATION, kafkaAssembly, List.of(POOL_CONTROLLERS, POOL_MIXED, POOL_BROKERS), Map.of(), KafkaVersionTestUtils.DEFAULT_KRAFT_VERSION_CHANGE, SHARED_ENV_PROVIDER);
+        KafkaCluster cluster = KafkaCluster.fromCrd(Reconciliation.DUMMY_RECONCILIATION, kafkaAssembly, pools, VERSIONS, KafkaVersionTestUtils.DEFAULT_KRAFT_VERSION_CHANGE, null, SHARED_ENV_PROVIDER);
+
+        // configurations that are in the exception list
+        assertThat(cluster.configuration.getConfiguration(), CoreMatchers.containsString("listener.name.tls-9093.connections.max.reauth.ms=1000"));
+        assertThat(cluster.configuration.getConfiguration(), CoreMatchers.containsString("listener.name.tls-9093.max.connections=1000"));
+        assertThat(cluster.configuration.getConfiguration(), CoreMatchers.containsString("listener.name.plain-9092.connections.max.reauth.ms=1000"));
+        assertThat(cluster.configuration.getConfiguration(), CoreMatchers.containsString("listener.name.plain-9092.max.connections=1000"));
+
+        // configuration that is not in the exception
+        assertThat(cluster.configuration.getConfiguration(), not(CoreMatchers.containsString("listener.name.plain-9092.connections.max.idle.ms")));
+        assertThat(cluster.configuration.getConfiguration(), not(CoreMatchers.containsString("listener.name.tls-9093.connections.max.idle.ms")));
+
+        // configuration for a listener that does not exist
+        assertThat(cluster.configuration.getConfiguration(), not(CoreMatchers.containsString("listener.name.does-not-exist.connections.max.reauth.ms")));
+
+        // configurations for internal interfaces
+        assertThat(cluster.configuration.getConfiguration(), not(CoreMatchers.containsString("listener.name.REPLICATION-9091.connections.max.reauth.ms")));
+        assertThat(cluster.configuration.getConfiguration(), not(CoreMatchers.containsString("listener.name.CONTROLPLANE-9090.max.connections")));
+    }
+
     @Test
     public void testCruiseControl() {
         Kafka kafkaAssembly = new KafkaBuilder(KAFKA)
diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaConfigurationTests.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaConfigurationTests.java
@@ -4,14 +4,20 @@
  */
 package io.strimzi.operator.cluster.model;
 
+import io.strimzi.api.kafka.model.kafka.listener.GenericKafkaListener;
+import io.strimzi.api.kafka.model.kafka.listener.GenericKafkaListenerBuilder;
+import io.strimzi.api.kafka.model.kafka.listener.KafkaListenerType;
 import io.strimzi.operator.cluster.KafkaVersionTestUtils;
 import io.strimzi.operator.common.Reconciliation;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
+
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
 import static java.util.Collections.singletonMap;
+import static org.hamcrest.CoreMatchers.hasItems;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -140,4 +146,19 @@ public void testRemoteStorageExpirationThreadPoolSize() {
         assertConfigError("remote.log.manager.expiration.thread.pool.size", "0", "remote.log.manager.expiration.thread.pool.size has value 0 which less than the minimum value 1");
         assertConfigError("remote.log.manager.expiration.thread.pool.size", "-5", "remote.log.manager.expiration.thread.pool.size has value -5 which less than the minimum value 1");
     }
+
+    @Test
+    public void testModifyWithListeners() {
+        GenericKafkaListener listener1 = new GenericKafkaListenerBuilder()
+                .withName("listener1")
+                .withPort(9900)
+                .withType(KafkaListenerType.INTERNAL)
+                .build();
+
+        List<String> modifiedExceptionList = KafkaConfiguration.modifyWithListeners(List.of(listener1));
+
+        assertThat(modifiedExceptionList, hasItems("listener.name.listener1-9900.connections.max.reauth.ms",
+                "listener.name.listener1-9900.max.connections",
+                "listener.name.listener1-9900.max.connection.creation.rate"));
+    }
 }
diff --git a/documentation/api/io.strimzi.api.kafka.model.kafka.KafkaClusterSpec.adoc b/documentation/api/io.strimzi.api.kafka.model.kafka.KafkaClusterSpec.adoc
@@ -84,6 +84,12 @@ All other supported options are forwarded to Kafka, including the following exce
 ** `controller.quorum.election.backoff.max.ms`
 ** `controller.quorum.election.timeout.ms`
 ** `controller.quorum.fetch.timeout.ms`
+* Listener properties (which are configured in the following format: `listener.name.listener1-9900.connections.max.reauth.ms` where `listener1-9900` is the listener name and port joined by `-`):
+** `connections.max.reauth.ms`
+** `max.connections`
+** `max.connections.per.ip`
+** `max.connections.per.ip.overrides`
+** `max.connection.creation.rate`
 
 [id='property-kafka-brokerRackInitImage-{context}']
 = Configuring rack awareness and init container images
diff --git a/documentation/modules/appendix_crds.adoc b/documentation/modules/appendix_crds.adoc
@@ -89,7 +89,7 @@ include::../api/io.strimzi.api.kafka.model.kafka.KafkaClusterSpec.adoc[leveloffs
 |Configures listeners to provide access to Kafka brokers.
 |config
 |map
-|Kafka broker config properties with the following prefixes cannot be set: listeners, advertised., broker., listener., host.name, port, inter.broker.listener.name, sasl., ssl., security., password., log.dir, zookeeper.connect, zookeeper.set.acl, zookeeper.ssl, zookeeper.clientCnxnSocket, authorizer., super.user, cruise.control.metrics.topic, cruise.control.metrics.reporter.bootstrap.servers, node.id, process.roles, controller., metadata.log.dir, zookeeper.metadata.migration.enable, client.quota.callback.static.kafka.admin., client.quota.callback.static.produce, client.quota.callback.static.fetch, client.quota.callback.static.storage.per.volume.limit.min.available., client.quota.callback.static.excluded.principal.name.list, prometheus.metrics.reporter. (with the exception of: zookeeper.connection.timeout.ms, sasl.server.max.receive.size, ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols, ssl.secure.random.implementation, cruise.control.metrics.topic.num.partitions, cruise.control.metrics.topic.replication.factor, cruise.control.metrics.topic.retention.ms, cruise.control.metrics.topic.auto.create.retries, cruise.control.metrics.topic.auto.create.timeout.ms, cruise.control.metrics.topic.min.insync.replicas, broker.session.timeout.ms, broker.heartbeat.interval.ms, controller.socket.timeout.ms, controller.quorum.election.backoff.max.ms, controller.quorum.election.timeout.ms, controller.quorum.fetch.timeout.ms).
+|Kafka broker config properties with certain prefixes cannot be set unless it is in the exception list. Consult the documentation for the list of forbidden prefixes and exceptions.
 |storage
 |xref:type-EphemeralStorage-{context}[`EphemeralStorage`], xref:type-PersistentClaimStorage-{context}[`PersistentClaimStorage`], xref:type-JbodStorage-{context}[`JbodStorage`]
 |**The `storage` property has been deprecated.** Use `KafkaNodePool` resources. Storage is now configured in the `KafkaNodePool` resources and this option is ignored.

Original file line number	Diff line number	Diff line change
`@@ -918,7 +918,6 @@ private void handleUserTaskStatusResponse(Reconciliation reconciliation, CruiseC`
`918`	`918`	`if (taskStatusJson.has(CruiseControlRebalanceKeys.LOAD_BEFORE_OPTIMIZATION.getKey()) &&`
`919`	`919`	`taskStatusJson.has(CruiseControlRebalanceKeys.LOAD_AFTER_OPTIMIZATION.getKey())) {`
`920`	`920`	`LOGGER.infoCr(reconciliation, "Rebalance ({}) optimization proposal is now ready", sessionId);`
`921`		`- System.out.println("Status" + taskStatusJson);`
`922`	`921`	`p.complete(buildRebalanceStatus(kafkaRebalance, sessionId, KafkaRebalanceState.ProposalReady, taskStatusJson, conditions));`
`923`	`922`	`break;`
`924`	`923`	`}`