fix(kraft): filter controller-only nodes from Cruise Control remove_broker

controllers/cruisecontroltask_controller.go

@@ -177,15 +177,30 @@ func (r *CruiseControlTaskReconciler) Reconcile(ctx context.Context, request ctr
 			brokerIDs = append(brokerIDs, task.BrokerID)
 		}
 
-		cruiseControlOpRef, err := r.removeBrokers(ctx, instance, operationTTLSecondsAfterFinished, brokerIDs)
-		if err != nil {
-			return requeueWithError(log, fmt.Sprintf("creating CruiseControlOperation for downscale has failed, brokerIDs: %s", brokerIDs), err)
+		if instance.Spec.KRaftMode {
+			// In KRaft mode, CC has no information about the controller-only nodes, so a remove_broker
+			// request for one would fail in CC. Controller-only nodes are downscaled by deleting their
+			// pod directly (see reconcileKafkaPodDelete) and never receive a GracefulDownscaleRequired
+			// state, so this filter is defense-in-depth to keep the three broker-id CC operations
+			// (add_broker, remove_broker, rebalance) symmetric.
+			brokerIDs, err = util.FilterControllerOnlyNodesWithBrokerStates(brokerIDs, instance.Spec, instance.Status.BrokersState)
+			if err != nil {
+				return requeueWithError(log, fmt.Sprintf("failed to filter out controller-only nodes from the Kafka cluster, "+
+					"clusterName: %s, clusterNamespace: %s", instance.GetName(), instance.GetNamespace()), err)
+			}
 		}
 
-		// map the CC broker removal operation with each broker status
-		for _, task := range tasksAndStates.GetActiveTasksByOp(banzaiv1alpha1.OperationRemoveBroker) {
-			task.SetCruiseControlOperationRef(cruiseControlOpRef)
-			task.SetStateScheduled()
+		if len(brokerIDs) != 0 {
+			cruiseControlOpRef, err := r.removeBrokers(ctx, instance, operationTTLSecondsAfterFinished, brokerIDs)
+			if err != nil {
+				return requeueWithError(log, fmt.Sprintf("creating CruiseControlOperation for downscale has failed, brokerIDs: %s", brokerIDs), err)
+			}
+
+			// map the CC broker removal operation with each broker status
+			for _, task := range tasksAndStates.GetActiveTasksByOp(banzaiv1alpha1.OperationRemoveBroker) {
+				task.SetCruiseControlOperationRef(cruiseControlOpRef)
+				task.SetStateScheduled()
+			}
 		}
 
 	case tasksAndStates.NumActiveTasksByOp(banzaiv1alpha1.OperationRemoveDisks) > 0:

pkg/util/util.go

@@ -211,18 +211,42 @@ func GetBrokerFromKafkaClusterSpec(brokerId string, spec v1beta1.KafkaClusterSpe
 }
 
 func FilterControllerOnlyNodes(brokerIDs []string, spec v1beta1.KafkaClusterSpec) ([]string, error) {
+	return filterControllerOnlyNodes(brokerIDs, spec, nil)
+}
+
+// FilterControllerOnlyNodesWithBrokerStates filters controller-only nodes while preserving removed brokers
+// whose active downscale task only exists in status.
+func FilterControllerOnlyNodesWithBrokerStates(brokerIDs []string, spec v1beta1.KafkaClusterSpec, brokerStates map[string]v1beta1.BrokerState) ([]string, error) {
+	return filterControllerOnlyNodes(brokerIDs, spec, brokerStates)
+}
+
+func filterControllerOnlyNodes(brokerIDs []string, spec v1beta1.KafkaClusterSpec, brokerStates map[string]v1beta1.BrokerState) ([]string, error) {
 	var filteredIDs []string
 	for _, brokerId := range brokerIDs {
 		broker := GetBrokerFromKafkaClusterSpec(brokerId, spec)
-		if broker != nil {
-			bConfig, err := broker.GetBrokerConfig(spec)
+		if broker == nil {
+			brokerState, ok := brokerStates[brokerId]
+			if !ok || brokerState.ConfigurationBackup == "" {
+				if brokerStates != nil {
+					filteredIDs = append(filteredIDs, brokerId)
+				}
+				continue
+			}
+
+			backupBroker, err := GetBrokerFromBrokerConfigurationBackup(brokerState.ConfigurationBackup)
 			if err != nil {
 				return nil, err
 			}
+			broker = &backupBroker
+		}
 
-			if !bConfig.IsControllerOnlyNode() {
-				filteredIDs = append(filteredIDs, strconv.Itoa(int(broker.Id)))
-			}
+		bConfig, err := broker.GetBrokerConfig(spec)
+		if err != nil {
+			return nil, err
+		}
+
+		if !bConfig.IsControllerOnlyNode() {
+			filteredIDs = append(filteredIDs, strconv.Itoa(int(broker.Id)))
 		}
 	}
 

pkg/util/util_test.go

@@ -722,6 +722,78 @@ func TestFilterControllerOnlyNodes(t *testing.T) {
 	}
 }
 
+func TestFilterControllerOnlyNodesWithBrokerStates(t *testing.T) {
+	removedBrokerConfigBackup, err := GzipAndBase64BrokerConfiguration(&v1beta1.Broker{
+		Id: 103,
+		BrokerConfig: &v1beta1.BrokerConfig{
+			Roles: []string{"broker"},
+		},
+	})
+	require.NoError(t, err)
+
+	removedControllerConfigBackup, err := GzipAndBase64BrokerConfiguration(&v1beta1.Broker{
+		Id: 4,
+		BrokerConfig: &v1beta1.BrokerConfig{
+			Roles: []string{"controller"},
+		},
+	})
+	require.NoError(t, err)
+
+	testCases := []struct {
+		testName                  string
+		kafkaClusterSpec          v1beta1.KafkaClusterSpec
+		brokerStates              map[string]v1beta1.BrokerState
+		allBrokerIDs              []string
+		expectedIDsAfterFiltering []string
+	}{
+		{
+			testName: "removed broker is classified from status backup and preserved",
+			kafkaClusterSpec: v1beta1.KafkaClusterSpec{
+				KRaftMode: true,
+				Brokers: []v1beta1.Broker{
+					{
+						Id: 100,
+						BrokerConfig: &v1beta1.BrokerConfig{
+							Roles: []string{"broker"},
+						},
+					},
+					{
+						Id: 0,
+						BrokerConfig: &v1beta1.BrokerConfig{
+							Roles: []string{"controller"},
+						},
+					},
+				},
+			},
+			brokerStates: map[string]v1beta1.BrokerState{
+				"103": {
+					ConfigurationBackup: removedBrokerConfigBackup,
+				},
+				"4": {
+					ConfigurationBackup: removedControllerConfigBackup,
+				},
+			},
+			allBrokerIDs:              []string{"100", "0", "103", "4"},
+			expectedIDsAfterFiltering: []string{"100", "103"},
+		},
+		{
+			testName:                  "removed broker without status backup is preserved",
+			kafkaClusterSpec:          v1beta1.KafkaClusterSpec{KRaftMode: true},
+			brokerStates:              map[string]v1beta1.BrokerState{"103": {}},
+			allBrokerIDs:              []string{"103"},
+			expectedIDsAfterFiltering: []string{"103"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.testName, func(t *testing.T) {
+			filteredIDs, err := FilterControllerOnlyNodesWithBrokerStates(tc.allBrokerIDs, tc.kafkaClusterSpec, tc.brokerStates)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedIDsAfterFiltering, filteredIDs)
+		})
+	}
+}
+
 func TestConstructEListenerLabelName(t *testing.T) {
 	tests := []struct {
 		ingressConfigName string