fix(cruisecontrol): sequence CC rolls around scaling operations (fixes #301)

controllers/cruisecontroloperation_controller.go

@@ -25,6 +25,7 @@ import (
 
 	"emperror.dev/errors"
 	"github.com/go-logr/logr"
+	appsv1 "k8s.io/api/apps/v1"
 	apiErrors "k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -224,6 +225,11 @@ func (r *CruiseControlOperationReconciler) Reconcile(ctx context.Context, reques
 		return requeueAfter(defaultRequeueIntervalInSeconds)
 	}
 
+	// Do not execute a broker add/remove while the Cruise Control Deployment is mid-rollout (see #301).
+	if result, handled, err := r.requeueIfCCDeploymentNotRolledOut(ctx, log, kafkaCluster, ccOperationExecution.CurrentTaskOperation()); handled {
+		return result, err
+	}
+
 	log.Info("executing Cruise Control task", "operation", ccOperationExecution.CurrentTaskOperation(), "parameters", ccOperationExecution.CurrentTaskParameters())
 	// Executing operation
 	cruseControlTaskResult, err := r.executeOperation(ctx, ccOperationExecution)
@@ -270,6 +276,71 @@ func (r *CruiseControlOperationReconciler) addFinalizer(ctx context.Context, cur
 	return nil
 }
 
+// requeueIfCCDeploymentNotRolledOut defers execution of a broker add_broker/remove_broker operation while
+// the Cruise Control Deployment is mid-rollout. A broker add/remove regenerates capacity.json, which (being
+// hashed into the CC pod template) rolls the CC Deployment; during that RollingUpdate two CC pods briefly
+// run behind one Service, so submitting/tracking the operation against a rolling CC makes the fresh pod lose
+// the in-memory task and reset its metric-sampling window, stalling the operation (see #301). Waiting for a
+// settled, single-Ready-replica Deployment also guarantees an add runs against a CC that has already loaded
+// the new broker's exact capacity from capacity.json (no dependency on capacity estimation). Stop-execution
+// is never gated. The returned bool reports whether the caller should return the (result, error) as-is.
+func (r *CruiseControlOperationReconciler) requeueIfCCDeploymentNotRolledOut(ctx context.Context, log logr.Logger,
+	kafkaCluster *banzaiv1beta1.KafkaCluster, op banzaiv1alpha1.CruiseControlTaskOperation) (ctrl.Result, bool, error) {
+	if op != banzaiv1alpha1.OperationAddBroker && op != banzaiv1alpha1.OperationRemoveBroker {
+		return ctrl.Result{}, false, nil
+	}
+
+	deployment := &appsv1.Deployment{}
+	key := client.ObjectKey{
+		Name:      fmt.Sprintf("%s-cruisecontrol", kafkaCluster.Name),
+		Namespace: kafkaCluster.Namespace,
+	}
+	if err := r.Get(ctx, key, deployment); err != nil {
+		if apiErrors.IsNotFound(err) {
+			// No Cruise Control Deployment: no rollout in progress to race with, so do not gate.
+			return ctrl.Result{}, false, nil
+		}
+		result, wErr := requeueWithError(log, "could not determine Cruise Control Deployment rollout state", err)
+		return result, true, wErr
+	}
+
+	if isDeploymentRolling(deployment) {
+		log.Info("requeue: Cruise Control Deployment is mid-rollout; deferring broker operation to avoid racing a CC restart", "operation", op)
+		result, _ := requeueAfter(defaultRequeueIntervalInSeconds)
+		return result, true, nil
+	}
+	return ctrl.Result{}, false, nil
+}
+
+// isDeploymentRolling reports whether a Deployment is actively in the middle of a rollout: a new pod
+// template has been applied but not yet observed by the Deployment controller, its pods have surged above
+// the desired count, or not all running replicas are the latest revision yet. It deliberately keys off
+// positive evidence of an in-progress rollout rather than "fully settled" so that a Deployment whose status
+// has never been populated (observedGeneration == 0, e.g. under envtest where no Deployment controller
+// runs) reads as NOT rolling and the check does not block. Initial CruiseControl availability is enforced
+// separately by CruiseControlStatus.IsReady; this gate only guards against submitting a broker operation
+// while an already-running CC is being re-rolled (e.g. by a capacity.json change).
+func isDeploymentRolling(deployment *appsv1.Deployment) bool {
+	specReplicas := int32(1)
+	if deployment.Spec.Replicas != nil {
+		specReplicas = *deployment.Spec.Replicas
+	}
+	s := deployment.Status
+	switch {
+	case s.ObservedGeneration != 0 && deployment.Generation > s.ObservedGeneration:
+		// A new pod template was applied but the Deployment controller has not observed it yet.
+		return true
+	case s.Replicas > specReplicas:
+		// RollingUpdate surge: an old-revision pod is still running alongside the new one.
+		return true
+	case s.UpdatedReplicas < s.Replicas:
+		// Not all running pods are the latest revision yet.
+		return true
+	default:
+		return false
+	}
+}
+
 func (r *CruiseControlOperationReconciler) executeOperation(ctx context.Context, ccOperationExecution *banzaiv1alpha1.CruiseControlOperation) (*scale.Result, error) {
 	var cruseControlTaskResult *scale.Result
 	var err error

controllers/cruisecontroloperation_controller_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/mock/gomock"
+	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -54,6 +55,48 @@ func createCCRetryExecutionOperation(createTime time.Time, id string, operation
 	}
 }
 
+func TestIsDeploymentRolling(t *testing.T) {
+	i32 := func(v int32) *int32 { return &v }
+	dep := func(generation, observedGeneration int64, specReplicas, replicas, updated int32) *appsv1.Deployment {
+		return &appsv1.Deployment{
+			ObjectMeta: v1.ObjectMeta{Generation: generation},
+			Spec:       appsv1.DeploymentSpec{Replicas: i32(specReplicas)},
+			Status: appsv1.DeploymentStatus{
+				ObservedGeneration: observedGeneration,
+				Replicas:           replicas,
+				UpdatedReplicas:    updated,
+			},
+		}
+	}
+
+	tests := []struct {
+		name string
+		d    *appsv1.Deployment
+		want bool
+	}{
+		{"settled single replica is not rolling", dep(3, 3, 1, 1, 1), false},
+		{"new pod template not yet observed is rolling", dep(4, 3, 1, 1, 1), true},
+		{"surge: an old-revision pod still present is rolling", dep(3, 3, 1, 2, 1), true},
+		{"not all running replicas updated yet is rolling", dep(3, 3, 1, 1, 0), true},
+		// envtest / no Deployment controller: status never populated (observedGeneration == 0). Must read
+		// as NOT rolling so the gate does not block where nothing rolls the Deployment.
+		{"unpopulated status (observedGeneration 0) is not rolling", dep(1, 0, 1, 0, 0), false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, isDeploymentRolling(tt.d))
+		})
+	}
+
+	t.Run("nil spec.replicas defaults to 1; settled is not rolling", func(t *testing.T) {
+		d := &appsv1.Deployment{
+			ObjectMeta: v1.ObjectMeta{Generation: 1},
+			Status:     appsv1.DeploymentStatus{ObservedGeneration: 1, Replicas: 1, UpdatedReplicas: 1},
+		}
+		assert.False(t, isDeploymentRolling(d))
+	})
+}
+
 func TestSortOperations(t *testing.T) {
 	timeNow := time.Now()
 	testCases := []struct {

pkg/resources/cruisecontrol/cruisecontrol.go

@@ -18,6 +18,7 @@ package cruisecontrol
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	"emperror.dev/errors"
 	"github.com/go-logr/logr"
@@ -112,7 +113,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 			}
 
 			var config *corev1.ConfigMap
-			if isBrokerDeletionInProgress(r.KafkaCluster.Status.BrokersState) {
+			if isBrokerDeletionInProgress(r.KafkaCluster.Status.BrokersState) || isBrokerRemovalPending(r.KafkaCluster) {
 				key := types.NamespacedName{
 					Name:      fmt.Sprintf(configAndVolumeNameTemplate, r.KafkaCluster.Name),
 					Namespace: r.KafkaCluster.Namespace,
@@ -201,3 +202,27 @@ func isBrokerDeletionInProgress(brokerState map[string]v1beta1.BrokerState) bool
 	}
 	return false
 }
+
+// isBrokerRemovalPending reports whether a broker that is still present in the status has already been
+// dropped from the spec - i.e. a removal that has not yet been marked as a Cruise Control downscale.
+//
+// During this window the operator must keep reusing the already-deployed capacity.json instead of
+// regenerating a fallback entry for the departing broker. Regenerating it changes capacity.json, which
+// (because capacity.json is hashed into the Cruise Control pod template) rolls the Cruise Control
+// Deployment and resets CC's metric-sampling window - keeping CC un-ready exactly when
+// reconcileKafkaPodDelete needs CC ready (via BrokersWithState) to mark the downscale. That chicken-and-egg
+// otherwise prevents the remove_broker operation from ever being created and stalls the removal (see #301).
+// Once the pod is gone and the broker is dropped from the status too, capacity.json shrinks with a single
+// harmless roll and no operation in flight.
+func isBrokerRemovalPending(kafkaCluster *v1beta1.KafkaCluster) bool {
+	specBrokerIDs := make(map[string]struct{}, len(kafkaCluster.Spec.Brokers))
+	for i := range kafkaCluster.Spec.Brokers {
+		specBrokerIDs[strconv.Itoa(int(kafkaCluster.Spec.Brokers[i].Id))] = struct{}{}
+	}
+	for brokerID := range kafkaCluster.Status.BrokersState {
+		if _, ok := specBrokerIDs[brokerID]; !ok {
+			return true
+		}
+	}
+	return false
+}

pkg/resources/cruisecontrol/cruisecontrol_test.go

@@ -0,0 +1,70 @@
+// Copyright 2025 Adobe. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cruisecontrol
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/banzaicloud/koperator/api/v1beta1"
+)
+
+func TestIsBrokerRemovalPending(t *testing.T) {
+	cluster := func(specIDs []int32, statusIDs []string) *v1beta1.KafkaCluster {
+		kc := &v1beta1.KafkaCluster{}
+		for _, id := range specIDs {
+			kc.Spec.Brokers = append(kc.Spec.Brokers, v1beta1.Broker{Id: id})
+		}
+		kc.Status.BrokersState = map[string]v1beta1.BrokerState{}
+		for _, id := range statusIDs {
+			kc.Status.BrokersState[id] = v1beta1.BrokerState{}
+		}
+		return kc
+	}
+
+	tests := []struct {
+		testName string
+		cluster  *v1beta1.KafkaCluster
+		expected bool
+	}{
+		{
+			testName: "steady state: every status broker is in the spec",
+			cluster:  cluster([]int32{0, 1, 2}, []string{"0", "1", "2"}),
+			expected: false,
+		},
+		{
+			testName: "removal pending: a status broker was dropped from the spec",
+			cluster:  cluster([]int32{0, 1, 2}, []string{"0", "1", "2", "6"}),
+			expected: true,
+		},
+		{
+			testName: "upscale in progress: a new spec broker not yet in status is NOT a removal",
+			cluster:  cluster([]int32{0, 1, 2, 6}, []string{"0", "1", "2"}),
+			expected: false,
+		},
+		{
+			testName: "empty status",
+			cluster:  cluster([]int32{0, 1, 2}, nil),
+			expected: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.testName, func(t *testing.T) {
+			require.Equal(t, test.expected, isBrokerRemovalPending(test.cluster))
+		})
+	}
+}

tests/e2e/koperator_suite_test.go

@@ -77,6 +77,7 @@ var _ = ginkgo.When("Testing e2e test altogether", ginkgo.Ordered, func() {
 	testInstallKafkaCluster("../../config/samples/kraft/simplekafkacluster_kraft.yaml")
 	testProduceConsumeInternal()
 	testJmxExporter()
+	testKRaftBrokerScaling()
 	testUninstallKafkaCluster()
 	testUninstall()
 	snapshotClusterAndCompare(snapshottedInfo)