[KafkaMigration] Updating common.go to properly check if usedForKafkaAdmin is set

hack/boilerplate/header.generated.txt

@@ -1,4 +1,4 @@
-Copyright 2024 Cisco Systems, Inc. and/or its affiliates
+Copyright 2025 Cisco Systems, Inc. and/or its affiliates
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

hack/boilerplate/header.go.generated.txt

@@ -1,5 +1,5 @@
 /*
-Copyright 2024 Cisco Systems, Inc. and/or its affiliates
+Copyright 2025 Cisco Systems, Inc. and/or its affiliates
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

pkg/util/kafka/common.go

@@ -191,10 +191,12 @@ func GetBootstrapServersService(cluster *v1beta1.KafkaCluster) (string, error) {
 
 // GetBrokerContainerPort return broker container port
 func GetBrokerContainerPort(cluster *v1beta1.KafkaCluster) (int32, error) {
+	usedForKafkaAdminSet := false
 	containerPort := int32(0)
 	for _, lc := range cluster.Spec.ListenersConfig.InternalListeners {
 		if lc.UsedForKafkaAdminCommunication { // Optional override to return a port from a different listener. Needed if b2b communication is on an external listener and and you want the koperator to interact with kafka over a different port.
 			containerPort = lc.ContainerPort
+			usedForKafkaAdminSet = true
 			break
 		}
 		if lc.UsedForInnerBrokerCommunication && !lc.UsedForControllerCommunication {
@@ -203,14 +205,16 @@ func GetBrokerContainerPort(cluster *v1beta1.KafkaCluster) (int32, error) {
 		}
 	}
 
-	for _, lc := range cluster.Spec.ListenersConfig.ExternalListeners {
-		if lc.UsedForKafkaAdminCommunication {
-			containerPort = lc.ContainerPort
-			break
-		}
-		if lc.UsedForInnerBrokerCommunication {
-			containerPort = lc.ContainerPort
-			break
+	if !usedForKafkaAdminSet {
+		for _, lc := range cluster.Spec.ListenersConfig.ExternalListeners {
+			if lc.UsedForKafkaAdminCommunication {
+				containerPort = lc.ContainerPort
+				break
+			}
+			if lc.UsedForInnerBrokerCommunication {
+				containerPort = lc.ContainerPort
+				break
+			}
 		}
 	}