From 0c897d8f2ae14d374bfb5ff88173d21faf3a42fc Mon Sep 17 00:00:00 2001 From: Adi Muraru Date: Mon, 10 Aug 2026 15:36:32 +0200 Subject: [PATCH] fix(kraft): match controller listener by exact name, not string prefix configureBrokerKRaftMode identified the controller listener among the "listeners" entries with `listener[:len(controllerListenerName)] == strings.ToUpper(controllerListenerName)`. This slice-prefix comparison has three failure modes, none guarded by validation: - Panic (slice out of range) when controllerListenerName is longer than a listener entry string (long controller name + short-named listener). - False match when one listener name is a prefix of another (e.g. "CONTROLLER" vs "CONTROLLER-EXTERNAL"): a broker-only node would silently drop the CONTROLLER-EXTERNAL listener. - When no controller listener is defined (controllerListenerName == ""), the empty prefix matched every entry, so broker-only nodes were assigned an empty `listeners=` (a broker listening nowhere). Replace it with isControllerListenerEntry, which compares the listener NAME segment (before "://") exactly and treats an empty controllerListenerName as "no match". For every valid KRaft config (an internal listener marked usedForControllerCommunication, distinct listener names) the output is unchanged. The two golden strings that changed belong to an intentionally invalid KRaft topology in TestGenerateBrokerConfigKRaftModeSSL (the "controller" listener is declared as an external listener, so no controller listener is identified). Their previous values encoded the buggy empty-prefix behavior; such configs will be rejected by the KRaft admission validation. Co-Authored-By: Claude Opus 4.8 (1M context) (cherry picked from commit ad8df0a03eea4b0dcbfcf1c7cadb10f0162ec92f) --- pkg/resources/kafka/configmap.go | 18 ++++++++- pkg/resources/kafka/configmap_test.go | 56 ++++++++++++++++++++++++++- 2 files changed, 70 insertions(+), 4 deletions(-) diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go index 9fc9d8b09..5ea2716af 100644 --- a/pkg/resources/kafka/configmap.go +++ b/pkg/resources/kafka/configmap.go @@ -220,7 +220,7 @@ func configureBrokerKRaftMode(bConfig *v1beta1.BrokerConfig, brokerID int32, kaf if bConfig.IsControllerOnlyNode() { // "listeners" configuration can only contain controller configuration when the node is a controller-only node for _, listener := range listenerConfig { - if listener[:len(controllerListenerName)] == strings.ToUpper(controllerListenerName) { + if isControllerListenerEntry(listener, controllerListenerName) { if err := config.Set(kafkautils.KafkaConfigListeners, listener); err != nil { log.Error(err, fmt.Sprintf(kafkautils.BrokerConfigErrorMsgTemplate, kafkautils.KafkaConfigListeners)) } @@ -231,7 +231,7 @@ func configureBrokerKRaftMode(bConfig *v1beta1.BrokerConfig, brokerID int32, kaf // "listeners" configuration cannot contain broker configuration when the node is a broker-only node var nonControllerListener []string for _, listener := range listenerConfig { - if listener[:len(controllerListenerName)] != strings.ToUpper(controllerListenerName) { + if !isControllerListenerEntry(listener, controllerListenerName) { nonControllerListener = append(nonControllerListener, listener) } } @@ -428,6 +428,20 @@ func generateControlPlaneListener(iListeners []v1beta1.InternalListenerConfig) s return controlPlaneListener } +// isControllerListenerEntry reports whether a "listeners" entry (formatted as "NAME://:PORT") +// belongs to the controller listener identified by controllerListenerName. +// It compares the listener NAME segment exactly rather than by string prefix, avoiding both +// the panic risk of slicing a shorter listener string and false matches when one listener name +// is a prefix of another (e.g. "CONTROLLER" vs "CONTROLLER-EXTERNAL"). Returns false when +// controllerListenerName is empty so that an unset controller listener never matches. +func isControllerListenerEntry(listenerEntry, controllerListenerName string) bool { + if controllerListenerName == "" { + return false + } + name, _, _ := strings.Cut(listenerEntry, "://") + return strings.EqualFold(name, controllerListenerName) +} + func generateListenerSpecificConfig(kcs *v1beta1.KafkaClusterSpec, serverPasses map[string]string, log logr.Logger) (*properties.Properties, map[int32]*properties.Properties, []string) { config := properties.NewProperties() brokerConfigs := make(map[int32]*properties.Properties) diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go index fbfde0049..7142bfb2b 100644 --- a/pkg/resources/kafka/configmap_test.go +++ b/pkg/resources/kafka/configmap_test.go @@ -1590,7 +1590,7 @@ listener.name.external.ssl.truststore.location=/var/run/secrets/java.io/keystore listener.name.external.ssl.truststore.password= listener.name.external.ssl.truststore.type=JKS listener.security.protocol.map=EXTERNAL:SSL,CONTROLLER:SSL -listeners= +listeners=EXTERNAL://:9092,CONTROLLER://:9093 log.dirs=/test-kafka-logs/kafka,/test-kafka-logs-0/kafka metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter node.id=0 @@ -1613,7 +1613,7 @@ listener.name.external.ssl.truststore.location=/var/run/secrets/java.io/keystore listener.name.external.ssl.truststore.password= listener.name.external.ssl.truststore.type=JKS listener.security.protocol.map=EXTERNAL:SSL,CONTROLLER:SSL -listeners=EXTERNAL://:9092 +listeners=EXTERNAL://:9092,CONTROLLER://:9093 log.dirs=/test-kafka-logs/kafka node.id=500 process.roles=controller @@ -1655,3 +1655,55 @@ process.roles=controller }) } } + +func TestIsControllerListenerEntry(t *testing.T) { + tests := []struct { + testName string + listenerEntry string + controllerListenerName string + expected bool + }{ + { + testName: "exact controller listener match", + listenerEntry: "CONTROLLER://:9093", + controllerListenerName: "CONTROLLER", + expected: true, + }, + { + testName: "case-insensitive match", + listenerEntry: "CONTROLLER://:9093", + controllerListenerName: "controller", + expected: true, + }, + { + testName: "prefix collision must not match (CONTROLLER vs CONTROLLER-EXTERNAL)", + listenerEntry: "CONTROLLER-EXTERNAL://:9094", + controllerListenerName: "CONTROLLER", + expected: false, + }, + { + testName: "non-controller listener does not match", + listenerEntry: "INTERNAL://:9092", + controllerListenerName: "CONTROLLER", + expected: false, + }, + { + testName: "empty controller listener name never matches", + listenerEntry: "INTERNAL://:9092", + controllerListenerName: "", + expected: false, + }, + { + testName: "controller name longer than the listener entry does not panic and does not match", + listenerEntry: "IB://:9092", + controllerListenerName: "CONTROLLERLISTENERWITHAVERYLONGNAME", + expected: false, + }, + } + + for _, test := range tests { + t.Run(test.testName, func(t *testing.T) { + require.Equal(t, test.expected, isControllerListenerEntry(test.listenerEntry, test.controllerListenerName)) + }) + } +}