[CORE-149029] Fix preferred affinity restart loop caused by atomic list merge semantics

go.mod

@@ -97,7 +97,7 @@ require (
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
-	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.18.6 // indirect
 	github.com/mattn/go-colorable v0.1.15 // indirect
 	github.com/mattn/go-isatty v0.0.22 // indirect

pkg/resources/kafka/kafka.go

@@ -957,7 +957,7 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo
 		desiredPod.Spec.Tolerations = uniqueTolerations
 	}
 	// Check if the resource actually updated or if labels match TaintedBrokersSelector
-	patchResult, err := patch.DefaultPatchMaker.Calculate(currentPod, desiredPod)
+	patchResult, err := patch.DefaultPatchMaker.Calculate(currentPod, desiredPod, ignorePreferredAffinities())
 	switch {
 	case err != nil:
 		log.Error(err, "could not match objects", "kind", desiredType)

pkg/resources/kafka/util.go

@@ -21,6 +21,9 @@ import (
 	"sort"
 
 	"github.com/google/uuid"
+	json "github.com/json-iterator/go"
+
+	"github.com/banzaicloud/k8s-objectmatcher/patch"
 
 	"github.com/banzaicloud/koperator/api/v1beta1"
 )
@@ -73,3 +76,56 @@ func generateRandomClusterID() string {
 	randomUUID := uuid.New()
 	return base64.URLEncoding.EncodeToString(randomUUID[:])
 }
+
+const (
+	affinityKeyPod     = "podAffinity"
+	affinityKeyAntiPod = "podAntiAffinity"
+	affinityKeyNode    = "nodeAffinity"
+)
+
+// ignorePreferredAffinities returns a CalculateOption that strips
+// preferredDuringSchedulingIgnoredDuringExecution from podAffinity,
+// podAntiAffinity and nodeAffinity on both sides of the diff.
+//
+// Those lists have no patchMergeKey in the Kubernetes API types, so the
+// strategic merge patch treats them atomically: any external addition (e.g.
+// from an admission controller or the node lifecycle controller) would be
+// removed on every reconcile. This mirrors how tolerations are handled and
+// keeps preferred affinities from triggering unnecessary rolling restarts.
+func ignorePreferredAffinities() patch.CalculateOption {
+	return func(current, modified []byte) ([]byte, []byte, error) {
+		var err error
+		if current, err = deletePreferredAffinities(current); err != nil {
+			return nil, nil, err
+		}
+		if modified, err = deletePreferredAffinities(modified); err != nil {
+			return nil, nil, err
+		}
+		return current, modified, nil
+	}
+}
+
+func deletePreferredAffinities(obj []byte) ([]byte, error) {
+	var pod map[string]interface{}
+	if err := json.Unmarshal(obj, &pod); err != nil {
+		return nil, err
+	}
+	spec, ok := pod["spec"].(map[string]interface{})
+	if !ok {
+		return obj, nil
+	}
+	affinity, ok := spec["affinity"].(map[string]interface{})
+	if !ok {
+		return obj, nil
+	}
+	for _, key := range []string{affinityKeyPod, affinityKeyAntiPod, affinityKeyNode} {
+		if a, ok := affinity[key].(map[string]interface{}); ok {
+			delete(a, "preferredDuringSchedulingIgnoredDuringExecution")
+		}
+	}
+	marshaled, err := json.ConfigCompatibleWithStandardLibrary.Marshal(pod)
+	if err != nil {
+		return nil, err
+	}
+	return marshaled, nil
+}