Koperator unattended kafkacluster cluster downscale improvements

api/v1beta1/common_types.go

@@ -122,6 +122,13 @@ func (r CruiseControlState) IsDownscale() bool {
 		r == GracefulDownscaleScheduled
 }
 
+// IsDownscaleStalled returns true when a downscale is neither progressing nor succeeded:
+// CompletedWithError (auto-retried under ErrorPolicyRetry, may loop) or Paused (needs manual resume).
+// Used to emit a single edge-triggered log when a broker enters a non-progressing state.
+func (r CruiseControlState) IsDownscaleStalled() bool {
+	return r == GracefulDownscaleCompletedWithError || r == GracefulDownscalePaused
+}
+
 // IsRunningState returns true if CruiseControlState indicates
 // that the CC operation is scheduled and in-progress
 func (r CruiseControlState) IsRunningState() bool {

controllers/cruisecontroltask_controller.go

@@ -109,7 +109,7 @@ func (r *CruiseControlTaskReconciler) Reconcile(ctx context.Context, request ctr
 	}
 
 	// Update task states with information from Cruise Control
-	updateActiveTasks(tasksAndStates, ccOperations)
+	updateActiveTasks(log, tasksAndStates, ccOperations)
 
 	if err = r.UpdateStatus(ctx, instance, tasksAndStates); err != nil {
 		return requeueWithError(log, "failed to update Kafka Cluster status", err)
@@ -171,19 +171,22 @@ func (r *CruiseControlTaskReconciler) Reconcile(ctx context.Context, request ctr
 			}
 		}
 	case tasksAndStates.NumActiveTasksByOp(banzaiv1alpha1.OperationRemoveBroker) > 0:
-		var removeTask *CruiseControlTask
+		brokerIDs := make([]string, 0)
+		// gather brokers that are marked for removal as a result of downscale operation
 		for _, task := range tasksAndStates.GetActiveTasksByOp(banzaiv1alpha1.OperationRemoveBroker) {
-			removeTask = task
-			break
+			brokerIDs = append(brokerIDs, task.BrokerID)
 		}
 
-		cruiseControlOpRef, err := r.removeBroker(ctx, instance, operationTTLSecondsAfterFinished, removeTask.BrokerID)
+		cruiseControlOpRef, err := r.removeBrokers(ctx, instance, operationTTLSecondsAfterFinished, brokerIDs)
 		if err != nil {
-			return requeueWithError(log, fmt.Sprintf("creating CruiseControlOperation for downscale has failed, brokerID: %s", removeTask.BrokerID), err)
+			return requeueWithError(log, fmt.Sprintf("creating CruiseControlOperation for downscale has failed, brokerIDs: %s", brokerIDs), err)
 		}
 
-		removeTask.SetCruiseControlOperationRef(cruiseControlOpRef)
-		removeTask.SetStateScheduled()
+		// 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:
 		brokerLogDirsToRemove := make(map[string][]string)
@@ -367,8 +370,8 @@ func (r *CruiseControlTaskReconciler) addBrokers(ctx context.Context, kafkaClust
 	return r.createCCOperation(ctx, kafkaCluster, banzaiv1alpha1.ErrorPolicyRetry, ttlSecondsAfterFinished, banzaiv1alpha1.OperationAddBroker, bokerIDs, false, nil)
 }
 
-func (r *CruiseControlTaskReconciler) removeBroker(ctx context.Context, kafkaCluster *banzaiv1beta1.KafkaCluster, ttlSecondsAfterFinished *int, brokerID string) (corev1.LocalObjectReference, error) {
-	return r.createCCOperation(ctx, kafkaCluster, banzaiv1alpha1.ErrorPolicyRetry, ttlSecondsAfterFinished, banzaiv1alpha1.OperationRemoveBroker, []string{brokerID}, false, nil)
+func (r *CruiseControlTaskReconciler) removeBrokers(ctx context.Context, kafkaCluster *banzaiv1beta1.KafkaCluster, ttlSecondsAfterFinished *int, brokerIDs []string) (corev1.LocalObjectReference, error) {
+	return r.createCCOperation(ctx, kafkaCluster, banzaiv1alpha1.ErrorPolicyRetry, ttlSecondsAfterFinished, banzaiv1alpha1.OperationRemoveBroker, brokerIDs, false, nil)
 }
 
 func (r *CruiseControlTaskReconciler) removeDisks(ctx context.Context, kafkaCluster *banzaiv1beta1.KafkaCluster, ttlSecondsAfterFinished *int, brokerIdsToRemovedLogDirs map[string][]string) (corev1.LocalObjectReference, error) {
@@ -655,7 +658,7 @@ func getActiveTasksFromCluster(instance *banzaiv1beta1.KafkaCluster) *CruiseCont
 
 // updateActiveTasks updates the state of the tasks from the CruiseControlTasksAndStates instance by getting their
 // status from CruiseControlOperation
-func updateActiveTasks(tasksAndStates *CruiseControlTasksAndStates, ccOperations []*banzaiv1alpha1.CruiseControlOperation) {
+func updateActiveTasks(log logr.Logger, tasksAndStates *CruiseControlTasksAndStates, ccOperations []*banzaiv1alpha1.CruiseControlOperation) {
 	ccOperationMap := make(map[string]*banzaiv1alpha1.CruiseControlOperation)
 	for i := range ccOperations {
 		ccOperationMap[ccOperations[i].Name] = ccOperations[i]
@@ -666,6 +669,16 @@ func updateActiveTasks(tasksAndStates *CruiseControlTasksAndStates, ccOperations
 			continue
 		}
 
+		prev := task.BrokerState
 		task.FromResult(ccOperationMap[task.CruiseControlOperationReference.Name])
+		if !prev.IsDownscaleStalled() && task.BrokerState.IsDownscaleStalled() {
+			if task.BrokerState == banzaiv1beta1.GracefulDownscalePaused {
+				log.Info("broker downscale paused; manual resume required, broker retained in external listener config",
+					"brokerID", task.BrokerID, "state", task.BrokerState)
+			} else {
+				log.Info("broker downscale not progressing; broker retained in external listener config",
+					"brokerID", task.BrokerID, "state", task.BrokerState)
+			}
+		}
 	}
 }

controllers/cruisecontroltask_controller_test.go

@@ -343,6 +343,17 @@ func TestCreateCCOperation(t *testing.T) {
 				assert.Equal(t, "true", params[scale.ParamExcludeRemoved])
 			},
 		},
+		{
+			operationType:      banzaiv1alpha1.OperationRemoveBroker,
+			brokerIDs:          []string{"1", "2", "3"},
+			isJBOD:             false,
+			brokerIdsToLogDirs: nil,
+			parameterCheck: func(t *testing.T, params map[string]string) {
+				assert.Equal(t, "1,2,3", params[scale.ParamBrokerID])
+				assert.Equal(t, "true", params[scale.ParamExcludeDemoted])
+				assert.Equal(t, "true", params[scale.ParamExcludeRemoved])
+			},
+		},
 		{
 			operationType: banzaiv1alpha1.OperationRemoveDisks,
 			brokerIDs:     []string{"1", "2"},

controllers/tests/cruisecontroltask_controller_test.go

@@ -24,6 +24,7 @@ import (
 	. "github.com/onsi/gomega"
 	"go.uber.org/mock/gomock"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	"github.com/banzaicloud/koperator/controllers/tests/mocks"
 
@@ -456,6 +457,137 @@ var _ = Describe("CruiseControlTaskReconciler", func() {
 			}, taskExtendedTimeoutDuration, reconcilePollingPeriod).Should(BeTrue())
 		})
 	})
+	When("multiple brokers are removed", Serial, func() {
+		JustBeforeEach(func(ctx SpecContext) {
+			kafkaClusterCCReconciler.ScaleFactory = mocks.NewMockScaleFactory(getScaleMockCCTask1())
+			err := util.RetryOnConflict(util.DefaultBackOffForConflict, func() error {
+				if err := k8sClient.Get(ctx, types.NamespacedName{
+					Name:      kafkaCluster.Name,
+					Namespace: kafkaCluster.Namespace,
+				}, kafkaCluster); err != nil {
+					return err
+				}
+
+				for _, id := range []string{"1", "2"} {
+					brokerState := kafkaCluster.Status.BrokersState[id]
+					brokerState.GracefulActionState.CruiseControlState = v1beta1.GracefulDownscaleRequired
+					kafkaCluster.Status.BrokersState[id] = brokerState
+				}
+				return k8sClient.Status().Update(ctx, kafkaCluster)
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+		It("should create exactly one remove_broker CruiseControlOperation for all brokers", func(ctx SpecContext) {
+			Eventually(ctx, func() bool {
+				err := k8sClient.Get(ctx, types.NamespacedName{
+					Name:      kafkaCluster.Name,
+					Namespace: kafkaCluster.Namespace,
+				}, kafkaCluster)
+				Expect(err).NotTo(HaveOccurred())
+
+				brokerState1, ok1 := kafkaCluster.Status.BrokersState["1"]
+				brokerState2, ok2 := kafkaCluster.Status.BrokersState["2"]
+				if !ok1 || !ok2 {
+					return false
+				}
+				if brokerState1.GracefulActionState.CruiseControlOperationReference == nil ||
+					brokerState2.GracefulActionState.CruiseControlOperationReference == nil {
+					return false
+				}
+
+				operationList := &v1alpha1.CruiseControlOperationList{}
+				err = k8sClient.List(ctx, operationList, client.ListOption(client.InNamespace(kafkaCluster.Namespace)))
+				Expect(err).NotTo(HaveOccurred())
+
+				if len(operationList.Items) != 1 {
+					return false
+				}
+				operation = &operationList.Items[0]
+				return operation.CurrentTaskOperation() == v1alpha1.OperationRemoveBroker &&
+					brokerState1.GracefulActionState.CruiseControlOperationReference.Name == operation.Name &&
+					brokerState2.GracefulActionState.CruiseControlOperationReference.Name == operation.Name &&
+					brokerState1.GracefulActionState.CruiseControlState == v1beta1.GracefulDownscaleScheduled &&
+					brokerState2.GracefulActionState.CruiseControlState == v1beta1.GracefulDownscaleScheduled
+			}, taskExtendedTimeoutDuration, reconcilePollingPeriod).Should(BeTrue())
+		})
+	})
+	When("one broker is already scheduled and another is newly required", Serial, func() {
+		var existingOpName string
+
+		JustBeforeEach(func(ctx SpecContext) {
+			kafkaClusterCCReconciler.ScaleFactory = mocks.NewMockScaleFactory(getScaleMockCCTask1())
+
+			// Pre-create a CC operation for broker "1" simulating a previous batch that is
+			// already scheduled. The owner reference must be set so updateActiveTasks includes
+			// it in ccOperationMap; without it, FromResult receives nil and marks the broker
+			// GracefulDownscaleSucceeded, clobbering the pre-existing state.
+			existingOp := generateCruiseControlOperation(
+				fmt.Sprintf("pre-existing-op-%v", count),
+				namespace,
+				kafkaClusterCRName,
+			)
+			err := controllerutil.SetControllerReference(kafkaCluster, &existingOp, kafkaClusterCCReconciler.Scheme)
+			Expect(err).NotTo(HaveOccurred())
+			err = k8sClient.Create(ctx, &existingOp)
+			Expect(err).NotTo(HaveOccurred())
+			existingOpName = existingOp.Name
+
+			err = util.RetryOnConflict(util.DefaultBackOffForConflict, func() error {
+				if err := k8sClient.Get(ctx, types.NamespacedName{
+					Name:      kafkaCluster.Name,
+					Namespace: kafkaCluster.Namespace,
+				}, kafkaCluster); err != nil {
+					return err
+				}
+
+				broker1State := kafkaCluster.Status.BrokersState["1"]
+				broker1State.GracefulActionState.CruiseControlState = v1beta1.GracefulDownscaleScheduled
+				broker1State.GracefulActionState.CruiseControlOperationReference = &corev1.LocalObjectReference{Name: existingOpName}
+				kafkaCluster.Status.BrokersState["1"] = broker1State
+
+				broker2State := kafkaCluster.Status.BrokersState["2"]
+				broker2State.GracefulActionState.CruiseControlState = v1beta1.GracefulDownscaleRequired
+				kafkaCluster.Status.BrokersState["2"] = broker2State
+
+				return k8sClient.Status().Update(ctx, kafkaCluster)
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("should create a new remove_broker operation for the required broker only, leaving the scheduled broker untouched", func(ctx SpecContext) {
+			Eventually(ctx, func() bool {
+				err := k8sClient.Get(ctx, types.NamespacedName{
+					Name:      kafkaCluster.Name,
+					Namespace: kafkaCluster.Namespace,
+				}, kafkaCluster)
+				Expect(err).NotTo(HaveOccurred())
+
+				broker1State, ok1 := kafkaCluster.Status.BrokersState["1"]
+				broker2State, ok2 := kafkaCluster.Status.BrokersState["2"]
+				if !ok1 || !ok2 {
+					return false
+				}
+				if broker2State.GracefulActionState.CruiseControlOperationReference == nil {
+					return false
+				}
+
+				operationList := &v1alpha1.CruiseControlOperationList{}
+				err = k8sClient.List(ctx, operationList, client.ListOption(client.InNamespace(kafkaCluster.Namespace)))
+				Expect(err).NotTo(HaveOccurred())
+
+				if len(operationList.Items) != 2 {
+					return false
+				}
+				operation = &operationList.Items[0]
+
+				newOpName := broker2State.GracefulActionState.CruiseControlOperationReference.Name
+				return broker1State.GracefulActionState.CruiseControlOperationReference.Name == existingOpName &&
+					broker1State.GracefulActionState.CruiseControlState == v1beta1.GracefulDownscaleScheduled &&
+					newOpName != existingOpName &&
+					broker2State.GracefulActionState.CruiseControlState == v1beta1.GracefulDownscaleScheduled
+			}, taskExtendedTimeoutDuration, reconcilePollingPeriod).Should(BeTrue())
+		})
+	})
 })
 
 func getScaleMockCCTask1() *mocks.MockCruiseControlScaler {

openspec/changes/downscale-improvements/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-05-13

openspec/changes/downscale-improvements/design.md

@@ -0,0 +1,81 @@
+## Context
+
+The koperator manages Kafka cluster lifecycle via a set of reconcilers. Downscale (broker removal) is handled in two phases:
+
+1. **KafkaCluster reconciler** (`pkg/resources/kafka/kafka.go`): Detects brokers removed from spec. Collects all such brokers in one pass and sets them all to `GracefulDownscaleRequired` via a single atomic `UpdateBrokerStatus` call.
+
+2. **CruiseControlTask reconciler** (`controllers/cruisecontroltask_controller.go`): Picks up brokers in `GracefulDownscaleRequired` state and submits CC operations. The `add_broker` path already batches all pending brokers into one CC operation. The `remove_broker` path does not — it picks only the first task and breaks.
+
+3. **External listener reconcilers** (`pkg/resources/envoy/`): Gate broker inclusion in envoy config on `ShouldIncludeBroker()`. This function returns `false` when `brokerConfig == nil` (broker not in spec), causing draining brokers to vanish from envoy listener config immediately upon spec removal. Note: `pkg/resources/contouringress` and `pkg/resources/nodeportexternalaccess` iterate `Spec.Brokers` directly and are not affected by this fix (see Non-Goals).
+
+**Key invariant:** `GetActiveTasksByOp(OperationRemoveBroker)` only returns brokers in `GracefulDownscaleRequired` state (via `IsRequired()` → `IsRequiredState()`). Brokers already `Scheduled` or `Running` are not returned. Because the KafkaCluster reconciler transitions all removed brokers atomically, all brokers from a single manifest apply are guaranteed to be in `Required` state simultaneously when the CC task reconciler fires.
+
+## Goals / Non-Goals
+
+**Goals:**
+- All broker IDs removed in a single manifest apply are submitted as one CC `remove_broker` operation.
+- Brokers removed from spec remain in external listener config until `GracefulDownscaleSucceeded`.
+- Brokers stuck in `GracefulDownscaleCompletedWithError` or `GracefulDownscalePaused` remain in external listener config (broker still holds data; manual investigation needed).
+
+**Non-Goals:**
+- Guaranteed single-operation batching across multiple sequential manifest applies (best-effort; second apply may produce a second CC operation if first batch is already `Scheduled`).
+- Changes to CRD schema or the CC REST API.
+- KRaft controller-only node handling (controller-only nodes skip CC graceful downscale entirely; unchanged).
+
+## Decisions
+
+### D1: Mirror `addBrokers` pattern for `removeBrokers`
+
+The `addBrokers` helper (lines 366-368) already accepts `[]string` and submits one CC operation. The `removeBroker` helper (lines 370-372) takes a single `string`.
+
+**Decision:** Rename `removeBroker` → `removeBrokers`, change signature to `[]string`, and replace the early-break loop with the collect-all pattern.
+
+**Why not a separate aggregation layer?** The batching boundary is already correct — `GetActiveTasksByOp` returns exactly the set to batch. No new abstraction needed.
+
+### D2: Fix `ShouldIncludeBroker` for envoy callers that enumerate status∪spec
+
+`ShouldIncludeBroker` is called by the envoy external listener reconcilers (configmap, service, deployment). When `brokerConfig == nil`, the function currently falls through to `return false`.
+
+**Real precondition for the fix:** The fallback only fires for reconcilers that enumerate brokers via `GetBrokerIdsFromStatusAndSpec` (status∪spec union), which means removed brokers are still visited after spec removal. Reconcilers that iterate `Spec.Brokers` directly (Contour, NodePort) never pass a removed broker to `ShouldIncludeBroker` in the first place — the fix is invisible to them. Any future listener reconciler must use `GetBrokerIdsFromStatusAndSpec` to benefit automatically; iterating `Spec.Brokers` silently bypasses this protection.
+
+**Decision:** Add a fallback block for `brokerConfig == nil`: check the broker's `CruiseControlState` in status. If `IsDownscale() && !IsSucceeded()` and the broker has the requested `ingressConfigName` in its `ExternalListenerConfigNames`, return `true`.
+
+**Why `ExternalListenerConfigNames` check?** A broker may have been associated with a specific ingress config. Re-using the persisted `ExternalListenerConfigNames` (set when the pod was created, never cleared until `GracefulDownscaleSucceeded`) ensures we only retain the broker for the listener configs it actually served.
+
+**Why `IsDownscale() && !IsSucceeded()` instead of an explicit state list?**
+`IsDownscale()` covers all 6 downscale states. Excluding `IsSucceeded()` retains the broker in all non-terminal states, including `CompletedWithError` and `Paused`, which is the desired behavior for manual investigation. If new downscale states are added to the enum in future, they're covered automatically.
+
+**Alternatives rejected:**
+- Fix each envoy caller individually: more code, same logic duplicated.
+- Add a new function: unnecessary indirection; `ShouldIncludeBroker` is the right seam for envoy callers.
+
+### D3: Task order to satisfy CI (green at every commit)
+
+The original plan placed failing tests before implementation. With CI gating on green builds, the order must be:
+
+```
+Commit 1: Unit test for createCCOperation (passes immediately — tests downstream, not the wrapper)
+Commit 2: Implement removeBrokers + integration test (both green together)
+Commit 3: Implement ShouldIncludeBroker fix + unit test (both green together)
+Commit 4: E2E test + sample manifest
+```
+
+## Risks / Trade-offs
+
+**[Race: second manifest apply before first batch starts]** → If a user applies a second spec change (removing more brokers) before the first CC operation is created, those new brokers will be included in the same batch (still in `Required` state). If the first CC operation is already `Scheduled`, new brokers get a separate operation. Acceptable; same behavior as `add_broker`.
+
+**[CompletedWithError brokers stay in envoy indefinitely]** → A broker that fails CC draining stays in external listener config. This is intentional (data is still present, clients need connectivity), but operators must monitor and manually recover. No change from current behavior for the envoy side — previously, the broker would have been dropped from envoy even with data present, which was worse.
+
+**[ExternalListenerConfigNames populated assumption]** → The fix assumes `ExternalListenerConfigNames` is non-empty for any broker that was ever reconciled. This field is set on pod creation and never cleared. Clusters created before this field existed would not benefit from the fix for those brokers, but all newly created or recently reconciled brokers are covered.
+
+## Migration Plan
+
+No migration required. Both fixes are backwards-compatible:
+- `removeBrokers` produces the same CC API calls as `removeBroker` for single-broker downscales.
+- `ShouldIncludeBroker` only changes behavior for brokers with `brokerConfig == nil` (already removed from spec); existing behavior for in-spec brokers is unchanged.
+
+Rollback: revert the two commits. No persistent state is affected.
+
+## Open Questions
+
+None — all decisions resolved during exploration.