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)) + }) + } +}