fix decom happening on multiple NPs concurrently

src/go/k8s/pkg/resources/statefulset_scale.go

@@ -73,8 +73,10 @@ const (
 func (r *StatefulSetResource) handleScaling(ctx context.Context) error {
 	log := r.logger.WithName("handleScaling").WithValues("nodepool", r.nodePool.Name)
 
+	// This is special - we allow only one decom at a time, ACROSS all nodePools. It's not per nodepool.
 	// if a decommission is already in progress, handle it first. If it's not finished, it will return an error
 	// which will requeue the reconciliation. We can't do any further scaling until it's finished.
+	// handleDecommissionInProgress is supposed to exit with error if a decom is already in progress, so we don't start another decom.
 	if err := r.handleDecommissionInProgress(ctx, log); err != nil {
 		return err
 	}
@@ -124,7 +126,6 @@ func (r *StatefulSetResource) handleScaling(ctx context.Context) error {
 	}
 
 	if npCurrentReplicas == 0 {
-		log.Info("We're done here!!!!!!!!!!!!!!!!111111111111111")
 		// we 're done here
 		return nil
 	}
@@ -177,7 +178,9 @@ func (r *StatefulSetResource) handleDecommissionInProgress(ctx context.Context,
 	_, err := strconv.Atoi(strings.TrimPrefix(brokerPod, r.LastObservedState.Name+"-")) // jb FIXME: use some better mechanism to get the pods for this pool.
 	if !strings.HasPrefix(brokerPod, r.LastObservedState.Name) || err != nil {          // Need to be really strict. If default nodepool is reconciled, the pod of a different nodepool would also have the prefix. Make sure, it's prefix + - + ordinal.
 		log.Info("broker is not part of this nodepool. Ignoring", "brokerID", brokerID, "pod name", brokerPod)
-		return nil
+
+		// We must return an error, or logic will continue to run, and we decom another broker, even if one is in progress already
+		return fmt.Errorf("decom is in progress, but it's part of this nodepool. erroring so this is retried. only one decom cluster-wide is allowed.")
 	}
 
 	if !r.nodePool.Removed && *r.nodePool.Replicas >= r.pandaCluster.Status.NodePools[r.LastObservedState.Name].CurrentReplicas { // NodePool name is NOT equal to STS name, or not supposed to be.