fix(kraft): match controller listener by exact name, not string prefix

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)

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