Add additional labels to brokers for backward compatibility while performing migration

Author: hvan

api/v1beta1/kafkacluster_types.go

@@ -1102,13 +1102,18 @@ func (bConfig *BrokerConfig) GetBrokerAnnotations() map[string]string {
 
 // GetBrokerLabels returns the labels that are applied to broker pods
 func (bConfig *BrokerConfig) GetBrokerLabels(kafkaClusterName string, brokerId int32, kRaftMode bool) map[string]string {
-	kraftLabels := make(map[string]string, 0)
+	var kraftLabels map[string]string
 	if kRaftMode {
 		kraftLabels = map[string]string{
 			ProcessRolesKey:     strings.Join(bConfig.Roles, "_"),
 			IsControllerNodeKey: fmt.Sprintf("%t", bConfig.IsControllerNode()),
 			IsBrokerNodeKey:     fmt.Sprintf("%t", bConfig.IsBrokerNode()),
 		}
+	} else { // in ZK mode -> new labels for backward compatibility for the headless service when going from ZK to KRaft
+		kraftLabels = map[string]string{
+			IsControllerNodeKey: fmt.Sprintf("%t", false),
+			IsBrokerNodeKey:     fmt.Sprintf("%t", true),
+		}
 	}
 	return util.MergeLabels(
 		bConfig.BrokerLabels,

api/v1beta1/kafkacluster_types_test.go

@@ -445,10 +445,12 @@ func TestGetBrokerLabels(t *testing.T) {
 		{
 			testName: "Labels in zookeeper mode",
 			expectedLabels: map[string]string{
-				AppLabelKey:      expectedDefaultLabelApp,
-				BrokerIdLabelKey: strconv.Itoa(expectedBrokerId),
-				KafkaCRLabelKey:  expectedKafkaCRName,
-				"test_label_key": "test_label_value",
+				AppLabelKey:         expectedDefaultLabelApp,
+				BrokerIdLabelKey:    strconv.Itoa(expectedBrokerId),
+				KafkaCRLabelKey:     expectedKafkaCRName,
+				IsBrokerNodeKey:     "true",
+				IsControllerNodeKey: "false",
+				"test_label_key":    "test_label_value",
 			},
 			brokerConfig: &BrokerConfig{
 				Roles: nil,

pkg/resources/kafka/configmap_test.go

@@ -1004,7 +1004,7 @@ process.roles=broker,controller
 `},
 		},
 		{
-			testName: "a Kafka cluster with a mix of broker-only, controller-only, and combined roles; controller nodes with multiple mount paths, and various migration configs set on each brokers",
+			testName: "a Kafka cluster with a mix of broker-only, controller-only, and combined roles; and various migration configs set on each brokers",
 			brokers: []v1beta1.Broker{
 				{
 					Id: 0,
@@ -1173,7 +1173,6 @@ process.roles=controller
 `,
 				`advertised.listeners=INTERNAL://kafka-100.kafka.svc.cluster.local:9092
 broker.id=100
-control.plane.listener.name=CONTROLLER
 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
@@ -1185,7 +1184,6 @@ zookeeper.connect=example.zk:2181/kafka
 `,
 				`advertised.listeners=INTERNAL://kafka-200.kafka.svc.cluster.local:9092
 broker.id=200
-control.plane.listener.name=CONTROLLER
 controller.listener.names=CONTROLLER
 [email protected]:9093,[email protected]:9093
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092