adding fix for koperator to add interBrokerListenerName one at a time and check for security.inter.broker.protocol (#101)

* adding fix for koperator to add interBrokerListenerName one at a time

* updating test to include inter.broker.listener.name

* fixing code that checks for inter.broker.listener.name

* removing unecessary function calls

* updating test expectations to be alphabetized

* adding generateListenerSpecificConfig back

* adding merges back in

* reverting back to having getConfigProperties only call generateListenerSpecificConfig

* removing unnecessary comments

---------

Co-authored-by: Cameron Wright <[email protected]>

Author: cawright-rh

pkg/resources/kafka/configmap.go

@@ -51,8 +51,16 @@ func (r *Reconciler) getConfigProperties(bConfig *v1beta1.BrokerConfig, broker v
 	// }
 
 	// Add listener configuration
-	listenerConf, _ := generateListenerSpecificConfig(&r.KafkaCluster.Spec, serverPasses, log)
-	config.Merge(listenerConf)
+	generalConfig, brokerConfigs, _ := generateListenerSpecificConfig(&r.KafkaCluster.Spec, serverPasses, log)
+
+	brokerConfig, exists := brokerConfigs[broker.Id]
+	if !exists {
+		log.Error(nil, fmt.Sprintf("No specific listener config found for broker %d, using default config", broker.Id))
+		brokerConfig = properties.NewProperties()
+	}
+
+	config.Merge(brokerConfig)
+	config.Merge(generalConfig)
 
 	// Cruise Control metrics reporter configuration
 	r.configCCMetricsReporter(broker, config, clientPass, log)
@@ -63,7 +71,7 @@ func (r *Reconciler) getConfigProperties(bConfig *v1beta1.BrokerConfig, broker v
 	if r.KafkaCluster.Spec.KRaftMode {
 		configureBrokerKRaftMode(bConfig, broker.Id, r.KafkaCluster, config, quorumVoters, serverPasses, extListenerStatuses, intListenerStatuses, log, brokerReadOnlyConfig)
 	} else {
-		configureBrokerZKMode(broker.Id, r.KafkaCluster, config, serverPasses, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, log)
+		configureBrokerZKMode(broker.Id, r.KafkaCluster, config, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, log)
 	}
 
 	// This logic prevents the removal of the mountPath from the broker configmap
@@ -197,9 +205,8 @@ func configureBrokerKRaftMode(bConfig *v1beta1.BrokerConfig, brokerID int32, kaf
 		}
 	}
 
-	// Add listener configuration
-	listenerConf, listenerConfig := generateListenerSpecificConfig(&kafkaCluster.Spec, serverPasses, log)
-	config.Merge(listenerConf)
+	// // Add listener configuration
+	_, _, listenerConfig := generateListenerSpecificConfig(&kafkaCluster.Spec, serverPasses, log)
 
 	var advertisedListenerConf []string
 	// only expose "advertised.listeners" when the node serves as a regular broker or a combined node
@@ -249,17 +256,12 @@ func shouldConfigureControllerQuorumForBroker(brokerReadOnlyConfig *properties.P
 	return !found || migrationBrokerControllerQuorumConfigEnabled.Value() == "true"
 }
 
-func configureBrokerZKMode(brokerID int32, kafkaCluster *v1beta1.KafkaCluster, config *properties.Properties,
-	serverPasses map[string]string, extListenerStatuses, intListenerStatuses,
+func configureBrokerZKMode(brokerID int32, kafkaCluster *v1beta1.KafkaCluster, config *properties.Properties, extListenerStatuses, intListenerStatuses,
 	controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList, log logr.Logger) {
 	if err := config.Set(kafkautils.KafkaConfigBrokerID, brokerID); err != nil {
 		log.Error(err, fmt.Sprintf(kafkautils.BrokerConfigErrorMsgTemplate, kafkautils.KafkaConfigBrokerID))
 	}
 
-	// Add listener configuration
-	listenerConf, _ := generateListenerSpecificConfig(&kafkaCluster.Spec, serverPasses, log)
-	config.Merge(listenerConf)
-
 	// Add advertised listener configuration
 	advertisedListenerConf := generateAdvertisedListenerConfig(brokerID, kafkaCluster.Spec.ListenersConfig,
 		extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses)
@@ -403,20 +405,12 @@ func generateControlPlaneListener(iListeners []v1beta1.InternalListenerConfig) s
 	return controlPlaneListener
 }
 
-func generateListenerSpecificConfig(kcs *v1beta1.KafkaClusterSpec, serverPasses map[string]string, log logr.Logger) (*properties.Properties, []string) {
+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)
 
 	l := kcs.ListenersConfig
-	r := kcs.ReadOnlyConfig
-
-	securityProtocolSet := false
-	for _, broker := range kcs.Brokers {
-		b := broker.ReadOnlyConfig
-		if strings.Contains(b, kafkautils.KafkaConfigSecurityInterBrokerProtocol+"=") {
-			securityProtocolSet = true
-			break
-		}
-	}
+	//r := kcs.ReadOnlyConfig
 
 	interBrokerListenerName, securityProtocolMapConfig, listenerConfig, internalListenerSSLConfig, externalListenerSSLConfig := getListenerSpecificConfig(&l, serverPasses, log)
 
@@ -436,19 +430,33 @@ func generateListenerSpecificConfig(kcs *v1beta1.KafkaClusterSpec, serverPasses
 		log.Error(err, fmt.Sprintf("setting '%s' parameter in broker configuration resulted an error", kafkautils.KafkaConfigListenerSecurityProtocolMap))
 	}
 
-	if !securityProtocolSet {
-		if !strings.Contains(r, kafkautils.KafkaConfigSecurityInterBrokerProtocol+"=") {
-			if err := config.Set(kafkautils.KafkaConfigInterBrokerListenerName, interBrokerListenerName); err != nil {
-				log.Error(err, fmt.Sprintf("setting '%s' parameter in broker configuration resulted an error", kafkautils.KafkaConfigInterBrokerListenerName))
+	for _, broker := range kcs.Brokers {
+		brokerConfig := properties.NewProperties()
+		brokerID := broker.Id
+
+		b := broker.ReadOnlyConfig
+		trimmedConfig := strings.TrimSpace(b)
+
+		if strings.Contains(trimmedConfig, kafkautils.KafkaConfigSecurityInterBrokerProtocol+"=") {
+			log.Info("Security InterBrokerProtocol is set for this broker, skipping config update", "broker", broker)
+		} else {
+			log.Info("Security InterBrokerProtocol NOT found for broker, setting inter.broker.listener.name",
+				"interBrokerListenerName", interBrokerListenerName, "broker", broker)
+
+			if err := brokerConfig.Set(kafkautils.KafkaConfigInterBrokerListenerName, interBrokerListenerName); err != nil {
+				log.Error(err, fmt.Sprintf("setting '%s' parameter in broker configuration resulted in an error",
+					kafkautils.KafkaConfigInterBrokerListenerName))
 			}
 		}
+
+		brokerConfigs[brokerID] = brokerConfig
 	}
 
 	if err := config.Set(kafkautils.KafkaConfigListeners, listenerConfig); err != nil {
 		log.Error(err, fmt.Sprintf("setting '%s' parameter in broker configuration resulted an error", kafkautils.KafkaConfigListeners))
 	}
 
-	return config, listenerConfig
+	return config, brokerConfigs, listenerConfig
 }
 
 func getListenerSpecificConfig(l *v1beta1.ListenersConfig, serverPasses map[string]string, log logr.Logger) (string, []string, []string, map[string]string, map[string]string) {

pkg/resources/kafka/configmap_test.go

@@ -619,6 +619,7 @@ zookeeper.connect=example.zk:2181/`,
 broker.id=0
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
+inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
 listeners=INTERNAL://:9092
 security.inter.broker.protocol=SASL_SSL