From 484385015d2565452df28b14b635cbb8a4d3fbbc Mon Sep 17 00:00:00 2001 From: Adi Muraru Date: Mon, 10 Aug 2026 15:38:11 +0200 Subject: [PATCH 1/2] fix(kraft): filter controller-only nodes from Cruise Control remove_broker The add_broker and rebalance branches of the Cruise Control task controller already drop controller-only nodes before calling Cruise Control (CC has no knowledge of them in KRaft mode), but the remove_broker branch did not. It relied entirely on the upstream invariant that controller-only nodes never receive a GracefulDownscaleRequired state. If that invariant ever regressed, a controller-only broker id would be sent to CC's remove_broker unfiltered and error out / stall the operation. Apply util.FilterControllerOnlyNodes in the remove_broker branch too, so all three broker-id CC operations are symmetrically guarded, and only issue the CC request when the filtered id list is non-empty. Co-Authored-By: Claude Opus 4.8 (1M context) (cherry picked from commit f626ded40d0c491f8e1519107afd9f572694bb21) --- controllers/cruisecontroltask_controller.go | 29 ++++++++++++++++----- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/controllers/cruisecontroltask_controller.go b/controllers/cruisecontroltask_controller.go index 9f7279c64..54fbd5b33 100644 --- a/controllers/cruisecontroltask_controller.go +++ b/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.FilterControllerOnlyNodes(brokerIDs, instance.Spec) + 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: From 46532c638af7086d803893dd97874e29babd457f Mon Sep 17 00:00:00 2001 From: Adi Muraru Date: Wed, 19 Aug 2026 15:01:50 +0200 Subject: [PATCH 2/2] fix(kraft): preserve removed brokers in CC downscale filter --- controllers/cruisecontroltask_controller.go | 2 +- pkg/util/util.go | 34 ++++++++-- pkg/util/util_test.go | 72 +++++++++++++++++++++ 3 files changed, 102 insertions(+), 6 deletions(-) diff --git a/controllers/cruisecontroltask_controller.go b/controllers/cruisecontroltask_controller.go index 54fbd5b33..d2092b9b6 100644 --- a/controllers/cruisecontroltask_controller.go +++ b/controllers/cruisecontroltask_controller.go @@ -183,7 +183,7 @@ func (r *CruiseControlTaskReconciler) Reconcile(ctx context.Context, request ctr // 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.FilterControllerOnlyNodes(brokerIDs, instance.Spec) + 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) diff --git a/pkg/util/util.go b/pkg/util/util.go index 87d8fecb6..cd2155fb7 100644 --- a/pkg/util/util.go +++ b/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))) } } diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go index 38c98f031..42e7c0e77 100644 --- a/pkg/util/util_test.go +++ b/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