Repository: kubernetes-sigs/kueue

Issue:
Refactor updating workload via PatchAdmissionStatus to make sure update is not saved on error
<!-- Please only use this template for submitting clean up requests -->

**What would you like to be cleaned**:

Calling PatchAdmissionStatus can currently lead to subtle bugs, when the function fails on error, because the object passed to the function is modified.

To achieve that we could have a new helper dedicated to Workloads which, say `workload.PatchStatus`:
1. takes an internal DeepCopy of the workload
2. calls PatchAdmissionStatus or another helper 
3. updates the workload pointer pass to invoke the function on success
4. discards the changes in case of failure

Alternatively, return the new version of the object on succeessful request (but this is more code changes). 


**Why is this needed**:

To avoid modifying the cached valued of the Workload in case the update fails. This inconsistency may result in subtle issues.

Discussion:
@mimowo: /assign @mbobrovskyi 
cc @mszadkow 
tentatively as a continuation of https://github.com/kubernetes-sigs/kueue/pull/7709

Solution patch:
diff --git a/cmd/experimental/kueue-populator/README.md b/cmd/experimental/kueue-populator/README.md
index 57e3f01a8..be90ef76b 100644
--- a/cmd/experimental/kueue-populator/README.md
+++ b/cmd/experimental/kueue-populator/README.md
@@ -35,6 +35,16 @@ The `kueue-populator` can be deployed to a Kubernetes cluster using the Kustomiz
     kubectl apply -k config
     ```
 
+### Installation via Helm
+
+You can also install the `kueue-populator` using the provided Helm chart.
+
+```bash
+helm install kueue-populator ./charts/kueue-populator --namespace kueue-system --create-namespace --wait
+```
+
+For more details on configuration and advanced usage, see the [Helm Chart README](charts/kueue-populator/README.md).
+
 ### Deploying to Kind with a Local Image
 
 To deploy the `kueue-populator` using a locally built image into a Kind cluster:
diff --git a/cmd/experimental/kueue-populator/charts/kueue-populator/README.md b/cmd/experimental/kueue-populator/charts/kueue-populator/README.md
index b721c112a..ca9b5cbd0 100644
--- a/cmd/experimental/kueue-populator/charts/kueue-populator/README.md
+++ b/cmd/experimental/kueue-populator/charts/kueue-populator/README.md
@@ -15,12 +15,20 @@ This Helm chart installs the Kueue Populator, a component designed to automatica
 -   [Helm](https://helm.sh/docs/intro/quickstart/#install-helm)
 -   Kubernetes cluster
 -   (Optional) [Cert-manager](https://cert-manager.io/docs/installation/)
--   Docker or a compatible container builder.
--   A container registry to push the image to.
+-   (Optional) Docker or a compatible container builder.
+-   (Optional) A container registry to push the image to.
+
+## Dependencies
+
+This chart depends on the [Kueue](https://github.com/kubernetes-sigs/kueue/tree/main/charts/kueue) chart.
+By default, it installs **Kueue v0.14.4**.
+
+-   If you already have Kueue installed, you can disable the dependency by setting `kueue.enabled=false`.
+-   The dependency is configured to enable `TopologyAwareScheduling` feature gate in Kueue.
 
 ## Building the Image
 
-You need to build and push the image to your own registry.
+If you want to make changes to the populator and deploy your own version, you need to build and push the image to your own registry.
  
 From the `cmd/experimental/kueue-populator` directory:
  
@@ -42,9 +50,21 @@ make image-build image-push IMAGE_REGISTRY=<YOUR_REGISTRY> GIT_TAG=latest
 
 ## Installation
 
-To install the chart, you MUST override the image repository and tag with the image you built and pushed.
+## Installation
+
+The chart is configured to use the official staging image by default.
+
+To install the chart with default settings:
+
+> The `--wait` flag is required to ensure that the Kueue controller and webhooks are fully ready before the chart attempts to create Kueue resources (like ClusterQueue) via post-install hooks. Without it, the installation may fail.
+
+```bash
+helm install kueue-populator ./charts/kueue-populator --namespace kueue-system --create-namespace --wait
+```
 
-The following commands assume you are in the `cmd/experimental/kueue-populator` directory.
+### Installing with a Custom Image
+
+If you built your own image, you **must** override the image repository and tag.
 
 Example using `--set`:
 
@@ -54,8 +74,6 @@ helm install kueue-populator ./charts/kueue-populator --namespace kueue-system -
   --set kueuePopulator.image.tag=latest
 ```
 
-> The `--wait` flag is required to ensure that the Kueue controller and webhooks are fully ready before the chart attempts to create Kueue resources (like ClusterQueue) via post-install hooks. Without it, the installation may fail.
-
 Example using a custom `my-values.yaml`:
 
 ```yaml
@@ -70,6 +88,39 @@ kueuePopulator:
 helm install kueue-populator ./charts/kueue-populator --namespace kueue-system --create-namespace --wait -f my-values.yaml
 ```
 
+### Installing with Topology and ResourceFlavor
+
+To enable Topology Aware Scheduling and configure the default ResourceFlavor with node labels, it is recommended to use a values file:
+
+```yaml
+# populator-config.yaml
+kueuePopulator:
+  config:
+    topology:
+      levels:
+        - nodeLabel: cloud.google.com/gke-nodepool
+    resourceFlavor:
+      nodeLabels:
+        cloud.google.com/gke-nodepool: "default-pool"
+```
+
+```bash
+helm install kueue-populator ./charts/kueue-populator --namespace kueue-system --create-namespace --wait -f populator-config.yaml
+```
+
+For simple configuration you may also use the minimalistic command:
+
+> When using `--set` for keys containing dots (e.g., `cloud.google.com/gke-nodepool`), you must escape the dots with a backslash.
+
+```bash
+helm install kueue-populator ./charts/kueue-populator \
+  --namespace kueue-system \
+  --create-namespace \
+  --wait \
+  --set kueuePopulator.config.topology.levels[0].nodeLabel="cloud\.google\.com/gke-nodepool" \
+  --set kueuePopulator.config.resourceFlavor.nodeLabels."cloud\.google\.com/gke-nodepool"=default-pool
+```
+
 ## Configuration
 
 ### Kueue Populator Configuration
diff --git a/cmd/experimental/kueue-populator/charts/kueue-populator/templates/setup-hook.yaml b/cmd/experimental/kueue-populator/charts/kueue-populator/templates/setup-hook.yaml
index 46fe5de79..2fa98f974 100644
--- a/cmd/experimental/kueue-populator/charts/kueue-populator/templates/setup-hook.yaml
+++ b/cmd/experimental/kueue-populator/charts/kueue-populator/templates/setup-hook.yaml
@@ -43,6 +43,7 @@ data:
       - key: "nvidia.com/gpu"
         operator: "Exists"
         effect: NoSchedule
+    ---
 {{- end }}
 {{- if .Values.kueuePopulator.config.clusterQueue }}
     apiVersion: kueue.x-k8s.io/v1beta1
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index fe537c4e6..cbec5bdbf 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -318,7 +318,7 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 			wl := e.Obj.DeepCopy()
 			if err := workload.PatchAdmissionStatus(ctx, s.client, wl, s.clock, func(wl *kueue.Workload) (bool, error) {
 				return workload.UnsetQuotaReservationWithCondition(wl, "Waiting", "waiting for all admitted workloads to be in PodsReady condition", s.clock.Now()), nil
-			}, workload.WithLooseOnApply()); err != nil {
+			}, workload.WithLooseOnApply(), workload.WithRetryOnConflictForPatch()); err != nil {
 				log.Error(err, "Could not update Workload status")
 			}
 			s.cache.WaitForPodsReady(ctx)
@@ -634,7 +634,7 @@ func (s *Scheduler) admit(ctx context.Context, e *entry, cq *schdcache.ClusterQu
 				wl.Status.UnhealthyNodes = nil
 			}
 			return true, nil
-		}, workload.WithLooseOnApply())
+		}, workload.WithLooseOnApply(), workload.WithRetryOnConflictForPatch())
 		if err == nil {
 			// Record metrics and events for quota reservation and admission
 			s.recordWorkloadAdmissionMetrics(newWorkload, e.Obj, admission)
@@ -799,7 +799,7 @@ func (s *Scheduler) requeueAndUpdate(ctx context.Context, e entry) {
 				updated = true
 			}
 			return updated, nil
-		}, workload.WithLooseOnApply()); err != nil {
+		}, workload.WithLooseOnApply(), workload.WithRetryOnConflictForPatch()); err != nil {
 			log.Error(err, "Could not update Workload status")
 		}
 		s.recorder.Eventf(e.Obj, corev1.EventTypeWarning, "Pending", api.TruncateEventMessage(e.inadmissibleMsg))
diff --git a/pkg/util/cert/cert.go b/pkg/util/cert/cert.go
index 813be4c34..38f7b8145 100644
--- a/pkg/util/cert/cert.go
+++ b/pkg/util/cert/cert.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
@@ -53,11 +54,16 @@ func BootstrapCerts(kubeConfig *rest.Config, cfg config.Configuration) error {
 			BindAddress:    "0",
 			FilterProvider: filters.WithAuthenticationAndAuthorization,
 		},
+		HealthProbeBindAddress: cfg.Health.HealthProbeBindAddress,
 	})
 	if err != nil {
 		return fmt.Errorf("unable to create bootstrap manager: %w", err)
 	}
 
+	if err := bootstrapMgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
+		return fmt.Errorf("unable to set up health check for bootstrap manager: %w", err)
+	}
+
 	certsReady := make(chan struct{})
 
 	// Add cert rotator to bootstrap manager using shared config.
diff --git a/pkg/util/client/client.go b/pkg/util/client/client.go
index c6d39a868..10e96fd8e 100644
--- a/pkg/util/client/client.go
+++ b/pkg/util/client/client.go
@@ -19,6 +19,7 @@ package client
 import (
 	"context"
 
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -43,7 +44,8 @@ type PatchOption func(*PatchOptions)
 // Typically, PatchOptions are constructed via DefaultPatchOptions and
 // modified using PatchOption functions (e.g., WithLoose).
 type PatchOptions struct {
-	Strict bool
+	Strict          bool
+	RetryOnConflict bool
 }
 
 // DefaultPatchOptions returns a new PatchOptions instance configured with
@@ -76,6 +78,13 @@ func WithLoose() PatchOption {
 	}
 }
 
+// WithRetryOnConflict is a PatchOption that enables retry on conflict when applying a patch.
+func WithRetryOnConflict() PatchOption {
+	return func(o *PatchOptions) {
+		o.RetryOnConflict = true
+	}
+}
+
 // Patch applies an update to a Kubernetes object using a patch-based workflow.
 //
 // The function first computes the "original" and "modified" states of the object
@@ -103,7 +112,7 @@ func WithLoose() PatchOption {
 //     This clears the ResourceVersion field in the original object to ensure
 //     it is always included in the generated patch.
 func Patch(ctx context.Context, c client.Client, obj client.Object, update UpdateFunc, options ...PatchOption) error {
-	return patchCommon(obj, update, func(patch client.Patch) error {
+	return patchCommon(ctx, c, obj, update, func(patch client.Patch) error {
 		return c.Patch(ctx, obj, patch)
 	}, options...)
 }
@@ -137,7 +146,7 @@ func Patch(ctx context.Context, c client.Client, obj client.Object, update Updat
 //     This clears the ResourceVersion field in the original object to ensure
 //     it is always included in the generated patch.
 func PatchStatus(ctx context.Context, c client.Client, obj client.Object, update UpdateFunc, options ...PatchOption) error {
-	return patchCommon(obj, update, func(patch client.Patch) error {
+	return patchCommon(ctx, c, obj, update, func(patch client.Patch) error {
 		return c.Status().Patch(ctx, obj, patch)
 	}, options...)
 }
@@ -146,11 +155,24 @@ func PatchStatus(ctx context.Context, c client.Client, obj client.Object, update
 // It returns an error if the patch could not be applied.
 type patchFunc func(patch client.Patch) error
 
-func patchCommon(obj client.Object, updateFn UpdateFunc, patchFn patchFunc, options ...PatchOption) error {
+func patchCommon(ctx context.Context, c client.Client, obj client.Object, updateFn UpdateFunc, patchFn patchFunc, options ...PatchOption) error {
 	opts := DefaultPatchOptions()
 	for _, opt := range options {
 		opt(opts)
 	}
+	if opts.RetryOnConflict {
+		fetch := false
+		objKey := client.ObjectKeyFromObject(obj)
+		return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+			if fetch {
+				if err := c.Get(ctx, objKey, obj); err != nil {
+					return err
+				}
+			}
+			fetch = true
+			return executePatch(obj, opts, updateFn, patchFn)
+		})
+	}
 	return executePatch(obj, opts, updateFn, patchFn)
 }
 
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 125826b53..ac3c7a112 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -965,8 +965,9 @@ type PatchAdmissionStatusOption func(*PatchAdmissionStatusOptions)
 // Typically, PatchAdmissionStatusOptions are constructed via DefaultPatchAdmissionStatusOptions and
 // modified using PatchAdmissionStatusOption functions (e.g., WithLoose).
 type PatchAdmissionStatusOptions struct {
-	StrictPatch bool
-	StrictApply bool
+	StrictPatch             bool
+	StrictApply             bool
+	RetryOnConflictForPatch bool
 }
 
 // DefaultPatchAdmissionStatusOptions returns a new PatchAdmissionStatusOptions instance configured with
@@ -1020,6 +1021,14 @@ func WithLooseOnApply() PatchAdmissionStatusOption {
 	}
 }
 
+// WithRetryOnConflictForPatch configures PatchAdmissionStatusOptions to enable retry logic on conflicts.
+// Note: This only works with merge patches.
+func WithRetryOnConflictForPatch() PatchAdmissionStatusOption {
+	return func(o *PatchAdmissionStatusOptions) {
+		o.RetryOnConflictForPatch = true
+	}
+}
+
 // PatchAdmissionStatus updates the admission status of a workload.
 // If the WorkloadRequestUseMergePatch feature is enabled, it uses a Merge Patch with update function.
 // Otherwise, it runs the update function and, if updated, applies the SSA Patch status.
@@ -1035,6 +1044,9 @@ func PatchAdmissionStatus(ctx context.Context, c client.Client, w *kueue.Workloa
 		if !opts.StrictPatch {
 			patchOptions = append(patchOptions, clientutil.WithLoose())
 		}
+		if opts.RetryOnConflictForPatch {
+			patchOptions = append(patchOptions, clientutil.WithRetryOnConflict())
+		}
 		err = clientutil.PatchStatus(ctx, c, wlCopy, func() (bool, error) {
 			return update(wlCopy)
 		}, patchOptions...)
diff --git a/site/content/en/docs/tasks/manage/setup_multikueue.md b/site/content/en/docs/tasks/manage/setup_multikueue.md
index c5be18d50..545244362 100644
--- a/site/content/en/docs/tasks/manage/setup_multikueue.md
+++ b/site/content/en/docs/tasks/manage/setup_multikueue.md
@@ -130,3 +130,127 @@ It provides a robust, modular, and extensible framework that helps other open so
 
 The integration with OCM is an optional solution that enables Kueue users to streamline the MultiKueue setup process, automate the generation of MultiKueue specific Kubeconfig, and enhance multicluster scheduling capabilities.
 For more details about this solution, please refer to this [link](https://github.com/open-cluster-management-io/ocm/tree/main/solutions/kueue-admission-check).
+
+## Setup MultiKueue with ClusterProfile API
+
+{{< feature-state state="alpha" for_version="v0.15" >}}
+
+The [ClusterProfile API](https://multicluster.sigs.k8s.io/concepts/cluster-profile-api/) provides a standardized, vendor-neutral interface for presenting cluster information. It allows defining cluster access information in a standardized `ClusterProfile` object and using credential plugins for authentication.
+
+### Enable MultiKueueClusterProfile feature gate
+Enable the `MultiKueueClusterProfile` feature gate. Refer to the
+[Installation guide](/docs/installation/#change-the-feature-gates-configuration)
+for instructions on configuring feature gates.
+
+### Create ClusterProfile objects
+
+If you are using a cloud provider, refer to the documentation on how to generate ClusterProfile objects (e.g. [GKE](https://docs.cloud.google.com/kubernetes-engine/fleet-management/docs/generate-inventory-for-integrations)). Alternatively, you can manually install the `ClusterProfile` CRD and objects for your clusters.
+
+To install the `ClusterProfile` CRD, run:
+```bash
+kubectl apply -f https://raw.githubusercontent.com/kubernetes-sigs/cluster-inventory-api/refs/heads/main/config/crd/bases/multicluster.x-k8s.io_clusterprofiles.yaml
+```
+
+To create a `ClusterProfile` object for `worker1-cluster`, run:
+```yaml
+apiVersion: multicluster.x-k8s.io/v1alpha1
+kind: ClusterProfile
+metadata:
+  name: worker1-cluster
+  namespace: kueue-system
+spec:
+  ...
+status:
+  accessProviders:
+  - name: ${PROVIDER_NAME}
+    cluster:
+      server: https://${SERVER_ENDPOINT}
+      certificate-authority-data: ${CERTIFICATE_AUTHORITY_DATA}
+```
+
+### Configure Kueue Manager
+
+Next, configure the controller manager config map with the credentials providers.
+
+```yaml
+apiVersion: v1
+data:
+  controller_manager_config.yaml: |
+    ...
+    multiKueue:
+      clusterProfile:
+        credentialsProviders:
+        - name: ${PROVIDER_NAME}
+          execConfig:
+            apiVersion: client.authentication.k8s.io/v1beta1
+            command: /plugins/${PLUGIN_COMMAND}
+kind: ConfigMap
+metadata:
+  labels:
+    app.kubernetes.io/name: kueue
+    control-plane: controller-manager
+  name: kueue-manager-config
+  namespace: kueue-system
+```
+
+### Install Required Plugins
+
+If your credentials provider requires an executable plugin, you must make it available to the Kueue manager.
+
+#### Add plugins via volume mounts
+
+You can use an `initContainer` to add the plugin to a shared `emptyDir` volume before the Kueue manager starts. The `kueue-controller-manager` container can then mount this volume to access the plugin.
+
+Here is an example patch for the `kueue-controller-manager` deployment that adds a custom authentication plugin:
+
+```yaml
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: kueue-controller-manager
+  namespace: kueue-system
+spec:
+  template:
+    spec:
+      initContainers:
+      - name: add-auth-plugin
+        image: ${PLUGIN_IMAGE}
+        command: ["cp", "${PLUGIN_COMMAND}", "/plugins/${PLUGIN_COMMAND}"]
+        volumeMounts:
+        - name: clusterprofile-plugins
+          mountPath: "/plugins"
+      containers:
+      - name: manager
+        volumeMounts:
+        - name: clusterprofile-plugins
+          mountPath: "/plugins"
+      volumes:
+      - name: clusterprofile-plugins
+        emptyDir: {}
+```
+
+This patch does the following:
+1.  Adds an `initContainer` that copies the `${PLUGIN_COMMAND}` from its container image to the `/plugins` directory in the shared volume.
+2.  Adds an `emptyDir` volume named `clusterprofile-plugins` to the pod.
+3.  Mounts the `clusterprofile-plugins` volume to the `manager` container, making the plugin available at `/plugins/${PLUGIN_COMMAND}`.
+
+#### Build a custom image
+
+Alternatively, you can build a custom Kueue manager image that includes your plugin. You would then update your Kueue deployment to use this new image.
+
+### Configure MultiKueueCluster objects
+
+When using the `ClusterProfile` API for authentication, configure your `MultiKueueCluster` objects to reference a `ClusterProfile` via the `clusterProfileRef` field, instead of providing `kubeconfig` directly.
+
+Here's an example `MultiKueueCluster` object using a `clusterProfileRef`:
+
+```yaml
+apiVersion: kueue.x-k8s.io/v1beta2
+kind: MultiKueueCluster
+metadata:
+  name: worker1-cluster
+spec:
+  clusterSource:
+    clusterProfileRef:
+      name: worker1-cluster
+```



Test patch:
diff --git a/pkg/controller/admissionchecks/multikueue/workload_test.go b/pkg/controller/admissionchecks/multikueue/workload_test.go
index 982fd5035..0117c7849 100644
--- a/pkg/controller/admissionchecks/multikueue/workload_test.go
+++ b/pkg/controller/admissionchecks/multikueue/workload_test.go
@@ -1450,163 +1450,174 @@ func TestWlReconcile(t *testing.T) {
 	}
 
 	for name, tc := range cases {
-		t.Run(name, func(t *testing.T) {
-			for feature, enabled := range tc.features {
-				features.SetFeatureGateDuringTest(t, feature, enabled)
-			}
-
-			ctx, _ := utiltesting.ContextWithLog(t)
-			managerBuilder := getClientBuilder(ctx)
-			managerBuilder = managerBuilder.WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
+		for _, useMergePatch := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s when the WorkloadRequestUseMergePatch feature is %t", name, useMergePatch), func(t *testing.T) {
+				features.SetFeatureGateDuringTest(t, features.WorkloadRequestUseMergePatch, useMergePatch)
 
-			workerClusters := []string{"worker1"}
-			if tc.useSecondWorker {
-				workerClusters = append(workerClusters, "worker2")
-			}
-			managerBuilder = managerBuilder.WithLists(&kueue.WorkloadList{Items: tc.managersWorkloads}, &batchv1.JobList{Items: tc.managersJobs})
-			managerBuilder = managerBuilder.WithStatusSubresource(slices.Map(tc.managersWorkloads, func(w *kueue.Workload) client.Object { return w })...)
-			managerBuilder = managerBuilder.WithStatusSubresource(slices.Map(tc.managersJobs, func(w *batchv1.Job) client.Object { return w })...)
-			managerBuilder = managerBuilder.WithObjects(
-				utiltestingapi.MakeMultiKueueConfig("config1").Clusters(workerClusters...).Obj(),
-				utiltestingapi.MakeAdmissionCheck("ac1").ControllerName(kueue.MultiKueueControllerName).
-					Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", "config1").
-					Obj(),
-			)
-
-			managerClient := managerBuilder.Build()
-			adapters, _ := jobframework.GetMultiKueueAdapters(sets.New("batch/job"))
-			cRec := newClustersReconciler(managerClient, TestNamespace, 0, defaultOrigin, nil, adapters, nil)
-
-			worker1Builder := getClientBuilder(ctx)
-			worker1Builder = worker1Builder.WithLists(&kueue.WorkloadList{Items: tc.worker1Workloads}, &batchv1.JobList{Items: tc.worker1Jobs})
-			worker1Client := worker1Builder.Build()
-
-			w1remoteClient := newRemoteClient(managerClient, nil, nil, defaultOrigin, "", adapters)
-			w1remoteClient.client = worker1Client
-			w1remoteClient.connecting.Store(false)
-			cRec.remoteClients["worker1"] = w1remoteClient
-
-			var worker2Client client.WithWatch
-			if tc.useSecondWorker {
-				worker2Builder := getClientBuilder(ctx)
-				worker2Builder = worker2Builder.WithLists(&kueue.WorkloadList{Items: tc.worker2Workloads}, &batchv1.JobList{Items: tc.worker2Jobs})
-				worker2Builder = worker2Builder.WithInterceptorFuncs(interceptor.Funcs{
-					Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
-						if tc.worker2OnGetError != nil {
-							return tc.worker2OnGetError
-						}
-						return c.Get(ctx, key, obj, opts...)
-					},
-					Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error {
-						if tc.worker2OnCreateError != nil {
-							return tc.worker2OnCreateError
-						}
-						return c.Create(ctx, obj, opts...)
-					},
-					Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error {
-						if tc.worker2OnDeleteError != nil {
-							return tc.worker2OnDeleteError
-						}
-						return c.Delete(ctx, obj, opts...)
-					},
-				})
-				worker2Client = worker2Builder.Build()
-
-				w2remoteClient := newRemoteClient(managerClient, nil, nil, defaultOrigin, "", adapters)
-				w2remoteClient.client = worker2Client
-				if !tc.worker2Reconnecting {
-					w2remoteClient.connecting.Store(false)
+				for feature, enabled := range tc.features {
+					features.SetFeatureGateDuringTest(t, feature, enabled)
 				}
-				cRec.remoteClients["worker2"] = w2remoteClient
-			}
 
-			helper, _ := admissioncheck.NewMultiKueueStoreHelper(managerClient)
-			recorder := &utiltesting.EventRecorder{}
-			mkDispatcherName := ptr.Deref(tc.dispatcherName, config.MultiKueueDispatcherModeAllAtOnce)
-			reconciler := newWlReconciler(managerClient, helper, cRec, defaultOrigin, recorder, defaultWorkerLostTimeout, time.Second, adapters, mkDispatcherName, WithClock(t, fakeClock))
+				ctx, _ := utiltesting.ContextWithLog(t)
+				managerBuilder := getClientBuilder(ctx)
+				managerBuilder = managerBuilder.WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
 
-			for _, val := range tc.managersDeletedWorkloads {
-				reconciler.Delete(event.DeleteEvent{
-					Object: val,
-				})
-			}
+				workerClusters := []string{"worker1"}
+				if tc.useSecondWorker {
+					workerClusters = append(workerClusters, "worker2")
+				}
+				managerBuilder = managerBuilder.WithLists(&kueue.WorkloadList{Items: tc.managersWorkloads}, &batchv1.JobList{Items: tc.managersJobs})
+				managerBuilder = managerBuilder.WithStatusSubresource(slices.Map(tc.managersWorkloads, func(w *kueue.Workload) client.Object { return w })...)
+				managerBuilder = managerBuilder.WithStatusSubresource(slices.Map(tc.managersJobs, func(w *batchv1.Job) client.Object { return w })...)
+				managerBuilder = managerBuilder.WithObjects(
+					utiltestingapi.MakeMultiKueueConfig("config1").Clusters(workerClusters...).Obj(),
+					utiltestingapi.MakeAdmissionCheck("ac1").ControllerName(kueue.MultiKueueControllerName).
+						Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", "config1").
+						Obj(),
+				)
 
-			_, gotErr := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: tc.reconcileFor, Namespace: TestNamespace}})
-			if diff := cmp.Diff(tc.wantError, gotErr, cmpopts.EquateErrors()); diff != "" {
-				t.Errorf("unexpected error (-want/+got):\n%s", diff)
-			}
+				managerClient := managerBuilder.Build()
+				adapters, _ := jobframework.GetMultiKueueAdapters(sets.New("batch/job"))
+				cRec := newClustersReconciler(managerClient, TestNamespace, 0, defaultOrigin, nil, adapters, nil)
+
+				worker1Builder := getClientBuilder(ctx)
+				worker1Builder = worker1Builder.WithLists(&kueue.WorkloadList{Items: tc.worker1Workloads}, &batchv1.JobList{Items: tc.worker1Jobs})
+				worker1Client := worker1Builder.Build()
+
+				w1remoteClient := newRemoteClient(managerClient, nil, nil, defaultOrigin, "", adapters)
+				w1remoteClient.client = worker1Client
+				w1remoteClient.connecting.Store(false)
+				cRec.remoteClients["worker1"] = w1remoteClient
+
+				var worker2Client client.WithWatch
+				if tc.useSecondWorker {
+					worker2Builder := getClientBuilder(ctx)
+					worker2Builder = worker2Builder.WithLists(&kueue.WorkloadList{Items: tc.worker2Workloads}, &batchv1.JobList{Items: tc.worker2Jobs})
+					worker2Builder = worker2Builder.WithInterceptorFuncs(interceptor.Funcs{
+						Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+							if tc.worker2OnGetError != nil {
+								return tc.worker2OnGetError
+							}
+							return c.Get(ctx, key, obj, opts...)
+						},
+						Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error {
+							if tc.worker2OnCreateError != nil {
+								return tc.worker2OnCreateError
+							}
+							return c.Create(ctx, obj, opts...)
+						},
+						Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error {
+							if tc.worker2OnDeleteError != nil {
+								return tc.worker2OnDeleteError
+							}
+							return c.Delete(ctx, obj, opts...)
+						},
+					})
+					worker2Client = worker2Builder.Build()
 
-			if diff := cmp.Diff(tc.wantEvents, recorder.RecordedEvents); diff != "" {
-				t.Errorf("unexpected events (-want/+got):\n%s", diff)
-			}
+					w2remoteClient := newRemoteClient(managerClient, nil, nil, defaultOrigin, "", adapters)
+					w2remoteClient.client = worker2Client
+					if !tc.worker2Reconnecting {
+						w2remoteClient.connecting.Store(false)
+					}
+					cRec.remoteClients["worker2"] = w2remoteClient
+				}
+
+				helper, _ := admissioncheck.NewMultiKueueStoreHelper(managerClient)
+				recorder := &utiltesting.EventRecorder{}
+				mkDispatcherName := ptr.Deref(tc.dispatcherName, config.MultiKueueDispatcherModeAllAtOnce)
+				reconciler := newWlReconciler(managerClient, helper, cRec, defaultOrigin, recorder, defaultWorkerLostTimeout, time.Second, adapters, mkDispatcherName, WithClock(t, fakeClock))
 
-			gotManagersWorkloads := &kueue.WorkloadList{}
-			if err := managerClient.List(ctx, gotManagersWorkloads); err != nil {
-				t.Errorf("unexpected list manager's workloads error: %s", err)
-			} else {
-				// ensure deterministic comparison
-				for i := range gotManagersWorkloads.Items {
-					sort.Strings(gotManagersWorkloads.Items[i].Status.NominatedClusterNames)
+				for _, val := range tc.managersDeletedWorkloads {
+					reconciler.Delete(event.DeleteEvent{
+						Object: val,
+					})
 				}
-				for i := range tc.wantManagersWorkloads {
-					sort.Strings(tc.wantManagersWorkloads[i].Status.NominatedClusterNames)
+
+				_, gotErr := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: tc.reconcileFor, Namespace: TestNamespace}})
+				if diff := cmp.Diff(tc.wantError, gotErr, cmpopts.EquateErrors()); diff != "" {
+					t.Errorf("unexpected error (-want/+got):\n%s", diff)
 				}
-				if diff := cmp.Diff(tc.wantManagersWorkloads, gotManagersWorkloads.Items, objCheckOpts...); diff != "" {
-					t.Errorf("unexpected manager's workloads (-want/+got):\n%s", diff)
+
+				if diff := cmp.Diff(tc.wantEvents, recorder.RecordedEvents); diff != "" {
+					t.Errorf("unexpected events (-want/+got):\n%s", diff)
 				}
-			}
 
-			gotWorker1Workloads := &kueue.WorkloadList{}
-			if err := worker1Client.List(ctx, gotWorker1Workloads); err != nil {
-				t.Errorf("unexpected list worker's workloads error: %s", err)
-			} else {
-				if diff := cmp.Diff(tc.wantWorker1Workloads, gotWorker1Workloads.Items, objCheckOpts...); diff != "" {
-					t.Errorf("unexpected worker's workloads (-want/+got):\n%s", diff)
+				// The fake client with patch.Apply cannot reset the Admission field (patch.Merge can).
+				// However, other important Status fields (e.g. Conditions) still reflect the change,
+				// so we deliberately ignore the Admission field here.
+				if features.Enabled(features.WorkloadRequestUseMergePatch) {
+					objCheckOpts = append(objCheckOpts, cmpopts.IgnoreFields(kueue.WorkloadStatus{}, "Admission"))
 				}
-			}
 
-			gotManagersJobs := &batchv1.JobList{}
-			if err := managerClient.List(ctx, gotManagersJobs); err != nil {
-				t.Errorf("unexpected list manager's jobs error %s", err)
-			} else {
-				if diff := cmp.Diff(tc.wantManagersJobs, gotManagersJobs.Items, objCheckOpts...); diff != "" {
-					t.Errorf("unexpected manager's jobs (-want/+got):\n%s", diff)
+				gotManagersWorkloads := &kueue.WorkloadList{}
+				if err := managerClient.List(ctx, gotManagersWorkloads); err != nil {
+					t.Errorf("unexpected list manager's workloads error: %s", err)
+				} else {
+					// ensure deterministic comparison
+					for i := range gotManagersWorkloads.Items {
+						sort.Strings(gotManagersWorkloads.Items[i].Status.NominatedClusterNames)
+					}
+					for i := range tc.wantManagersWorkloads {
+						sort.Strings(tc.wantManagersWorkloads[i].Status.NominatedClusterNames)
+					}
+					if diff := cmp.Diff(tc.wantManagersWorkloads, gotManagersWorkloads.Items, objCheckOpts...); diff != "" {
+						t.Errorf("unexpected manager's workloads (-want/+got):\n%s", diff)
+					}
 				}
-			}
 
-			gotWorker1Jobs := &batchv1.JobList{}
-			if err := worker1Client.List(ctx, gotWorker1Jobs); err != nil {
-				t.Error("unexpected list worker's jobs error")
-			} else {
-				if diff := cmp.Diff(tc.wantWorker1Jobs, gotWorker1Jobs.Items, objCheckOpts...); diff != "" {
-					t.Errorf("unexpected worker's jobs (-want/+got):\n%s", diff)
+				gotWorker1Workloads := &kueue.WorkloadList{}
+				if err := worker1Client.List(ctx, gotWorker1Workloads); err != nil {
+					t.Errorf("unexpected list worker's workloads error: %s", err)
+				} else {
+					if diff := cmp.Diff(tc.wantWorker1Workloads, gotWorker1Workloads.Items, objCheckOpts...); diff != "" {
+						t.Errorf("unexpected worker's workloads (-want/+got):\n%s", diff)
+					}
 				}
-			}
 
-			if tc.useSecondWorker {
-				gotWorker2Workloads := &kueue.WorkloadList{}
-				if err := worker2Client.List(ctx, gotWorker2Workloads); err != nil {
-					t.Errorf("unexpected list worker2 workloads error: %s", err)
+				gotManagersJobs := &batchv1.JobList{}
+				if err := managerClient.List(ctx, gotManagersJobs); err != nil {
+					t.Errorf("unexpected list manager's jobs error %s", err)
 				} else {
-					if diff := cmp.Diff(tc.wantWorker2Workloads, gotWorker2Workloads.Items, objCheckOpts...); diff != "" {
-						t.Errorf("unexpected worker2 workloads (-want/+got):\n%s", diff)
+					if diff := cmp.Diff(tc.wantManagersJobs, gotManagersJobs.Items, objCheckOpts...); diff != "" {
+						t.Errorf("unexpected manager's jobs (-want/+got):\n%s", diff)
 					}
 				}
 
-				gotWorker2Jobs := &batchv1.JobList{}
-				if err := worker2Client.List(ctx, gotWorker2Jobs); err != nil {
-					t.Errorf("unexpected list worker2 jobs error: %s", err)
+				gotWorker1Jobs := &batchv1.JobList{}
+				if err := worker1Client.List(ctx, gotWorker1Jobs); err != nil {
+					t.Error("unexpected list worker's jobs error")
 				} else {
-					if diff := cmp.Diff(tc.wantWorker2Jobs, gotWorker2Jobs.Items, objCheckOpts...); diff != "" {
-						t.Errorf("unexpected worker2 jobs (-want/+got):\n%s", diff)
+					if diff := cmp.Diff(tc.wantWorker1Jobs, gotWorker1Jobs.Items, objCheckOpts...); diff != "" {
+						t.Errorf("unexpected worker's jobs (-want/+got):\n%s", diff)
 					}
 				}
-			}
 
-			if l := reconciler.deletedWlCache.Len(); l > 0 {
-				t.Errorf("unexpected deletedWlCache len %d expecting 0", l)
-			}
-		})
+				if tc.useSecondWorker {
+					gotWorker2Workloads := &kueue.WorkloadList{}
+					if err := worker2Client.List(ctx, gotWorker2Workloads); err != nil {
+						t.Errorf("unexpected list worker2 workloads error: %s", err)
+					} else {
+						if diff := cmp.Diff(tc.wantWorker2Workloads, gotWorker2Workloads.Items, objCheckOpts...); diff != "" {
+							t.Errorf("unexpected worker2 workloads (-want/+got):\n%s", diff)
+						}
+					}
+
+					gotWorker2Jobs := &batchv1.JobList{}
+					if err := worker2Client.List(ctx, gotWorker2Jobs); err != nil {
+						t.Errorf("unexpected list worker2 jobs error: %s", err)
+					} else {
+						if diff := cmp.Diff(tc.wantWorker2Jobs, gotWorker2Jobs.Items, objCheckOpts...); diff != "" {
+							t.Errorf("unexpected worker2 jobs (-want/+got):\n%s", diff)
+						}
+					}
+				}
+
+				if l := reconciler.deletedWlCache.Len(); l > 0 {
+					t.Errorf("unexpected deletedWlCache len %d expecting 0", l)
+				}
+			})
+		}
 	}
 }
 
diff --git a/pkg/controller/jobs/job/job_controller_test.go b/pkg/controller/jobs/job/job_controller_test.go
index 220a6f71b..fc389e82d 100644
--- a/pkg/controller/jobs/job/job_controller_test.go
+++ b/pkg/controller/jobs/job/job_controller_test.go
@@ -4040,8 +4040,9 @@ func TestReconciler(t *testing.T) {
 					wlCheckOpts = workloadCmpOptsWithOwner
 				}
 
-				// Fake client with patch.Apply can't reset Admission field, patch.Merge can
-				// However other key Status fields indicate that change e.g. Conditions, thuse we choose to ignore the Admission field
+				// The fake client with patch.Apply cannot reset the Admission field (patch.Merge can).
+				// However, other important Status fields (e.g. Conditions) still reflect the change,
+				// so we deliberately ignore the Admission field here.
 				if features.Enabled(features.WorkloadRequestUseMergePatch) {
 					wlCheckOpts = append(wlCheckOpts, cmpopts.IgnoreFields(kueue.WorkloadStatus{}, "Admission"))
 				}
diff --git a/pkg/controller/jobs/pod/pod_controller_test.go b/pkg/controller/jobs/pod/pod_controller_test.go
index 11de068c9..c59ea8c4c 100644
--- a/pkg/controller/jobs/pod/pod_controller_test.go
+++ b/pkg/controller/jobs/pod/pod_controller_test.go
@@ -5728,8 +5728,9 @@ func TestReconciler(t *testing.T) {
 					t.Fatalf("Could not get Workloads after reconcile: %v", err)
 				}
 
-				// Fake client with patch.Apply can't reset Admission field, patch.Merge can
-				// However other key Status fields indicate that change e.g. Conditions, thuse we choose to ignore the Admission field
+				// The fake client with patch.Apply cannot reset the Admission field (patch.Merge can).
+				// However, other important Status fields (e.g. Conditions) still reflect the change,
+				// so we deliberately ignore the Admission field here.
 				if features.Enabled(features.WorkloadRequestUseMergePatch) {
 					tc.workloadCmpOpts = append(tc.workloadCmpOpts, cmpopts.IgnoreFields(kueue.WorkloadStatus{}, "Admission"))
 				}
@@ -5753,11 +5754,6 @@ func TestReconciler(t *testing.T) {
 
 func TestReconciler_ErrorFinalizingPod(t *testing.T) {
 	now := time.Now().Truncate(time.Second)
-	ctx, _ := utiltesting.ContextWithLog(t)
-	clientBuilder := utiltesting.NewClientBuilder()
-	if err := SetupIndexes(ctx, utiltesting.AsIndexer(clientBuilder)); err != nil {
-		t.Fatalf("Could not setup indexes: %v", err)
-	}
 
 	basePodWrapper := testingpod.MakePod("pod", "ns").
 		UID("test-uid").
@@ -5765,101 +5761,19 @@ func TestReconciler_ErrorFinalizingPod(t *testing.T) {
 		Request(corev1.ResourceCPU, "1").
 		Image("", nil)
 
-	pod := *basePodWrapper.
-		Clone().
-		ManagedByKueueLabel().
-		KueueFinalizer().
-		StatusPhase(corev1.PodSucceeded).
-		StatusMessage("Job finished successfully").
-		Obj()
-
-	wl := *utiltestingapi.MakeWorkload("unit-test", "ns").Finalizers(kueue.ResourceInUseFinalizerName).
+	baseWl := *utiltestingapi.MakeWorkload("unit-test", "ns").
 		PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
-		ReserveQuotaAt(utiltestingapi.MakeAdmission("cq").PodSets(utiltestingapi.MakePodSetAssignment(kueue.DefaultPodSetName).Obj()).Obj(), now).
-		AdmittedAt(true, now).
-		Obj()
+		ReserveQuotaAt(utiltestingapi.MakeAdmission("cq").PodSets(utiltestingapi.MakePodSetAssignment(kueue.DefaultPodSetName).Obj()).Obj(), now)
 
-	reqcount := 0
-	errMock := fmt.Errorf("connection refused: %w", syscall.ECONNREFUSED)
-
-	kcBuilder := clientBuilder.
-		WithObjects(utiltesting.MakeNamespace("ns")).
-		WithObjects(&pod).
-		WithStatusSubresource(&wl).
-		WithInterceptorFuncs(interceptor.Funcs{
-			Patch: func(ctx context.Context, client client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
-				_, isPod := obj.(*corev1.Pod)
-				if isPod {
-					defer func() { reqcount++ }()
-					if reqcount == 0 {
-						// return a connection refused error for the first update request.
-						return errMock
-					}
-					if reqcount == 1 {
-						// Exec a regular update operation for the second request
-						return client.Patch(ctx, obj, patch, opts...)
-					}
-				}
-				return client.Patch(ctx, obj, patch, opts...)
-			},
-			SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge,
-		})
-
-	kClient := kcBuilder.Build()
-	if err := ctrl.SetControllerReference(&pod, &wl, kClient.Scheme()); err != nil {
-		t.Fatalf("Could not setup owner reference in Workloads: %v", err)
-	}
-	if err := kClient.Create(ctx, &wl); err != nil {
-		t.Fatalf("Could not create workload: %v", err)
-	}
-	recorder := record.NewBroadcaster().NewRecorder(kClient.Scheme(), corev1.EventSource{Component: "test"})
-
-	reconciler, err := NewReconciler(ctx, kClient, nil, recorder, jobframework.WithClock(testingclock.NewFakeClock(now)))
-	if err != nil {
-		t.Errorf("Error creating the reconciler: %v", err)
-	}
-
-	podKey := client.ObjectKeyFromObject(&pod)
-	_, err = reconciler.Reconcile(ctx, reconcile.Request{
-		NamespacedName: podKey,
-	})
-
-	if diff := cmp.Diff(errMock, err, cmpopts.EquateErrors()); diff != "" {
-		t.Errorf("Expected reconcile error (-want,+got):\n%s", diff)
-	}
-
-	// Reconcile for the second time
-	_, err = reconciler.Reconcile(ctx, reconcile.Request{
-		NamespacedName: podKey,
-	})
-	if err != nil {
-		t.Errorf("Got unexpected error while running reconcile:\n%v", err)
-	}
-
-	var gotPod corev1.Pod
-	if err := kClient.Get(ctx, podKey, &gotPod); err != nil {
-		t.Fatalf("Could not get Pod after second reconcile: %v", err)
-	}
-	// Validate that pod has no finalizer after the second reconcile
 	wantPod := *basePodWrapper.
 		Clone().
 		ManagedByKueueLabel().
 		StatusPhase(corev1.PodSucceeded).
 		StatusMessage("Job finished successfully").
 		Obj()
-	if diff := cmp.Diff(wantPod, gotPod, podCmpOpts...); diff != "" {
-		t.Errorf("Pod after second reconcile (-want,+got):\n%s", diff)
-	}
-
-	var gotWorkloads kueue.WorkloadList
-	if err := kClient.List(ctx, &gotWorkloads); err != nil {
-		t.Fatalf("Could not get Workloads after second reconcile: %v", err)
-	}
 
 	// Workload should be finished after the second reconcile
-	wantWl := *utiltestingapi.MakeWorkload("unit-test", "ns").
-		PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
-		ReserveQuotaAt(utiltestingapi.MakeAdmission("cq").PodSets(utiltestingapi.MakePodSetAssignment(kueue.DefaultPodSetName).Obj()).Obj(), now).
+	wantWl := baseWl.Clone().
 		ControllerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 		Condition(
 			metav1.Condition{
@@ -5887,8 +5801,110 @@ func TestReconciler_ErrorFinalizingPod(t *testing.T) {
 		).
 		PastAdmittedTime(0).
 		Obj()
-	if diff := cmp.Diff([]kueue.Workload{wantWl}, gotWorkloads.Items, defaultWorkloadCmpOpts...); diff != "" {
-		t.Errorf("Workloads after second reconcile (-want,+got):\n%s", diff)
+
+	for _, useMergePatch := range []bool{false, true} {
+		t.Run(fmt.Sprintf("WorkloadRequestUseMergePatch enabled: %t", useMergePatch), func(t *testing.T) {
+			features.SetFeatureGateDuringTest(t, features.WorkloadRequestUseMergePatch, useMergePatch)
+
+			ctx, _ := utiltesting.ContextWithLog(t)
+			clientBuilder := utiltesting.NewClientBuilder()
+			if err := SetupIndexes(ctx, utiltesting.AsIndexer(clientBuilder)); err != nil {
+				t.Fatalf("Could not setup indexes: %v", err)
+			}
+
+			pod := *basePodWrapper.
+				Clone().
+				ManagedByKueueLabel().
+				KueueFinalizer().
+				StatusPhase(corev1.PodSucceeded).
+				StatusMessage("Job finished successfully").
+				Obj()
+
+			wl := *baseWl.Clone().AdmittedAt(true, now).Obj()
+
+			reqcount := 0
+			errMock := fmt.Errorf("connection refused: %w", syscall.ECONNREFUSED)
+
+			kcBuilder := clientBuilder.
+				WithObjects(utiltesting.MakeNamespace("ns")).
+				WithObjects(pod.DeepCopy()).
+				WithStatusSubresource(&kueue.Workload{}).
+				WithInterceptorFuncs(interceptor.Funcs{
+					Patch: func(ctx context.Context, client client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
+						_, isPod := obj.(*corev1.Pod)
+						if isPod {
+							defer func() { reqcount++ }()
+							if reqcount == 0 {
+								// return a connection refused error for the first update request.
+								return errMock
+							}
+							if reqcount == 1 {
+								// Exec a regular update operation for the second request
+								return client.Patch(ctx, obj, patch, opts...)
+							}
+						}
+						return client.Patch(ctx, obj, patch, opts...)
+					},
+					SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge,
+				})
+
+			kClient := kcBuilder.Build()
+			if err := ctrl.SetControllerReference(&pod, &wl, kClient.Scheme()); err != nil {
+				t.Fatalf("Could not setup owner reference in Workloads: %v", err)
+			}
+			if err := kClient.Create(ctx, &wl); err != nil {
+				t.Fatalf("Could not create workload: %v", err)
+			}
+			recorder := record.NewBroadcaster().NewRecorder(kClient.Scheme(), corev1.EventSource{Component: "test"})
+
+			reconciler, err := NewReconciler(ctx, kClient, nil, recorder, jobframework.WithClock(testingclock.NewFakeClock(now)))
+			if err != nil {
+				t.Errorf("Error creating the reconciler: %v", err)
+			}
+
+			podKey := client.ObjectKeyFromObject(&pod)
+			_, err = reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: podKey,
+			})
+
+			if diff := cmp.Diff(errMock, err, cmpopts.EquateErrors()); diff != "" {
+				t.Errorf("Expected reconcile error (-want,+got):\n%s", diff)
+			}
+
+			// Reconcile for the second time
+			_, err = reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: podKey,
+			})
+			if err != nil {
+				t.Errorf("Got unexpected error while running reconcile:\n%v", err)
+			}
+
+			var gotPod corev1.Pod
+			if err := kClient.Get(ctx, podKey, &gotPod); err != nil {
+				t.Fatalf("Could not get Pod after second reconcile: %v", err)
+			}
+
+			// Validate that pod has no finalizer after the second reconcile
+			if diff := cmp.Diff(wantPod, gotPod, podCmpOpts...); diff != "" {
+				t.Errorf("Pod after second reconcile (-want,+got):\n%s", diff)
+			}
+
+			var gotWorkloads kueue.WorkloadList
+			if err := kClient.List(ctx, &gotWorkloads); err != nil {
+				t.Fatalf("Could not get Workloads after second reconcile: %v", err)
+			}
+
+			// The fake client with patch.Apply cannot reset the Admission field (patch.Merge can).
+			// However, other important Status fields (e.g. Conditions) still reflect the change,
+			// so we deliberately ignore the Admission field here.
+			if features.Enabled(features.WorkloadRequestUseMergePatch) {
+				defaultWorkloadCmpOpts = append(defaultWorkloadCmpOpts, cmpopts.IgnoreFields(kueue.WorkloadStatus{}, "Admission"))
+			}
+
+			if diff := cmp.Diff([]kueue.Workload{*wantWl}, gotWorkloads.Items, defaultWorkloadCmpOpts...); diff != "" {
+				t.Errorf("Workloads after second reconcile (-want,+got):\n%s", diff)
+			}
+		})
 	}
 }
 
diff --git a/pkg/controller/jobs/raycluster/raycluster_controller_test.go b/pkg/controller/jobs/raycluster/raycluster_controller_test.go
index 5f7e837bd..f1b0b67be 100644
--- a/pkg/controller/jobs/raycluster/raycluster_controller_test.go
+++ b/pkg/controller/jobs/raycluster/raycluster_controller_test.go
@@ -731,8 +731,9 @@ func TestReconciler(t *testing.T) {
 				if err := kClient.List(ctx, &gotWorkloads); err != nil {
 					t.Fatalf("Could not get Workloads after reconcile: %v", err)
 				}
-				// Fake client with patch.Apply can't reset Admission field, patch.Merge can
-				// However other key Status fields indicate that change e.g. Conditions, thuse we choose to ignore the Admission field
+				// The fake client with patch.Apply cannot reset the Admission field (patch.Merge can).
+				// However, other important Status fields (e.g. Conditions) still reflect the change,
+				// so we deliberately ignore the Admission field here.
 				wlCheckOpts := workloadCmpOpts
 				if features.Enabled(features.WorkloadRequestUseMergePatch) {
 					wlCheckOpts = append(wlCheckOpts, cmpopts.IgnoreFields(kueue.WorkloadStatus{}, "Admission"))
diff --git a/pkg/scheduler/preemption/preemption_hierarchical_test.go b/pkg/scheduler/preemption/preemption_hierarchical_test.go
index 05bcb41d7..8969dd062 100644
--- a/pkg/scheduler/preemption/preemption_hierarchical_test.go
+++ b/pkg/scheduler/preemption/preemption_hierarchical_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package preemption
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -34,6 +35,7 @@ import (
 	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
 	"sigs.k8s.io/kueue/pkg/constants"
 	controllerconstants "sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/scheduler/flavorassigner"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	utiltestingapi "sigs.k8s.io/kueue/pkg/util/testing/v1beta2"
@@ -1813,77 +1815,82 @@ func TestHierarchicalPreemptions(t *testing.T) {
 		},
 	}
 	for name, tc := range cases {
-		t.Run(name, func(t *testing.T) {
-			ctx, log := utiltesting.ContextWithLog(t)
-			cl := utiltesting.NewClientBuilder().
-				WithLists(&kueue.WorkloadList{Items: tc.admitted}).
-				WithStatusSubresource(&kueue.Workload{}).
-				WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge}).
-				Build()
+		for _, useMergePatch := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s when the WorkloadRequestUseMergePatch feature is %t", name, useMergePatch), func(t *testing.T) {
+				features.SetFeatureGateDuringTest(t, features.WorkloadRequestUseMergePatch, useMergePatch)
 
-			cqCache := schdcache.New(cl)
-			for _, flv := range flavors {
-				cqCache.AddOrUpdateResourceFlavor(log, flv)
-			}
-			for _, cq := range tc.clusterQueues {
-				if err := cqCache.AddClusterQueue(ctx, cq); err != nil {
-					t.Fatalf("Couldn't add ClusterQueue to cache: %v", err)
+				ctx, log := utiltesting.ContextWithLog(t)
+				cl := utiltesting.NewClientBuilder().
+					WithLists(&kueue.WorkloadList{Items: tc.admitted}).
+					WithStatusSubresource(&kueue.Workload{}).
+					WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge}).
+					Build()
+
+				cqCache := schdcache.New(cl)
+				for _, flv := range flavors {
+					cqCache.AddOrUpdateResourceFlavor(log, flv)
+				}
+				for _, cq := range tc.clusterQueues {
+					if err := cqCache.AddClusterQueue(ctx, cq); err != nil {
+						t.Fatalf("Couldn't add ClusterQueue to cache: %v", err)
+					}
 				}
-			}
-			for _, cohort := range tc.cohorts {
-				if err := cqCache.AddOrUpdateCohort(cohort); err != nil {
-					t.Fatalf("Couldn't add Cohort to cache: %v", err)
+				for _, cohort := range tc.cohorts {
+					if err := cqCache.AddOrUpdateCohort(cohort); err != nil {
+						t.Fatalf("Couldn't add Cohort to cache: %v", err)
+					}
 				}
-			}
 
-			broadcaster := record.NewBroadcaster()
-			scheme := runtime.NewScheme()
-			if err := kueue.AddToScheme(scheme); err != nil {
-				t.Fatalf("Failed adding kueue scheme: %v", err)
-			}
-			recorder := broadcaster.NewRecorder(scheme, corev1.EventSource{Component: constants.AdmissionName})
-			preemptor := New(cl, workload.Ordering{}, recorder, nil, false, clocktesting.NewFakeClock(now))
+				broadcaster := record.NewBroadcaster()
+				scheme := runtime.NewScheme()
+				if err := kueue.AddToScheme(scheme); err != nil {
+					t.Fatalf("Failed adding kueue scheme: %v", err)
+				}
+				recorder := broadcaster.NewRecorder(scheme, corev1.EventSource{Component: constants.AdmissionName})
+				preemptor := New(cl, workload.Ordering{}, recorder, nil, false, clocktesting.NewFakeClock(now))
 
-			beforeSnapshot, err := cqCache.Snapshot(ctx)
-			if err != nil {
-				t.Fatalf("unexpected error while building snapshot: %v", err)
-			}
-			// make a working copy of the snapshotWorkingCopy than preemption can temporarily modify
-			snapshotWorkingCopy, err := cqCache.Snapshot(ctx)
-			if err != nil {
-				t.Fatalf("unexpected error while building snapshot: %v", err)
-			}
-			wlInfo := workload.NewInfo(tc.incoming)
-			wlInfo.ClusterQueue = tc.targetCQ
-			targets := preemptor.GetTargets(log, *wlInfo, tc.assignment, snapshotWorkingCopy)
-			preempted, failed, err := preemptor.IssuePreemptions(ctx, wlInfo, targets, snapshotWorkingCopy.ClusterQueue(wlInfo.ClusterQueue))
-			if err != nil {
-				t.Fatalf("Failed doing preemption")
-			}
-			if preempted != tc.wantPreempted {
-				t.Errorf("Reported %d preemptions, want %d", preempted, tc.wantPreempted)
-			}
-			if failed != 0 {
-				t.Errorf("Reported %d failed preemptions, want 0", failed)
-			}
+				beforeSnapshot, err := cqCache.Snapshot(ctx)
+				if err != nil {
+					t.Fatalf("unexpected error while building snapshot: %v", err)
+				}
+				// make a working copy of the snapshotWorkingCopy than preemption can temporarily modify
+				snapshotWorkingCopy, err := cqCache.Snapshot(ctx)
+				if err != nil {
+					t.Fatalf("unexpected error while building snapshot: %v", err)
+				}
+				wlInfo := workload.NewInfo(tc.incoming)
+				wlInfo.ClusterQueue = tc.targetCQ
+				targets := preemptor.GetTargets(log, *wlInfo, tc.assignment, snapshotWorkingCopy)
+				preempted, failed, err := preemptor.IssuePreemptions(ctx, wlInfo, targets, snapshotWorkingCopy.ClusterQueue(wlInfo.ClusterQueue))
+				if err != nil {
+					t.Fatalf("Failed doing preemption")
+				}
+				if preempted != tc.wantPreempted {
+					t.Errorf("Reported %d preemptions, want %d", preempted, tc.wantPreempted)
+				}
+				if failed != 0 {
+					t.Errorf("Reported %d failed preemptions, want 0", failed)
+				}
 
-			workloads := &kueue.WorkloadList{}
-			err = cl.List(ctx, workloads)
-			if err != nil {
-				t.Fatalf("Failed to List workloads: %v", err)
-			}
+				workloads := &kueue.WorkloadList{}
+				err = cl.List(ctx, workloads)
+				if err != nil {
+					t.Fatalf("Failed to List workloads: %v", err)
+				}
 
-			defaultCmpOpts := cmp.Options{
-				cmpopts.EquateEmpty(),
-				cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
-			}
-			if diff := cmp.Diff(tc.wantWorkloads, workloads.Items, defaultCmpOpts); diff != "" {
-				t.Errorf("Unexpected workloads (-want/+got)\n%s", diff)
-			}
+				defaultCmpOpts := cmp.Options{
+					cmpopts.EquateEmpty(),
+					cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
+					cmpopts.SortSlices(func(a, b metav1.Condition) bool { return a.Type < b.Type }),
+				}
+				if diff := cmp.Diff(tc.wantWorkloads, workloads.Items, defaultCmpOpts); diff != "" {
+					t.Errorf("Unexpected workloads (-want/+got)\n%s", diff)
+				}
 
-			if diff := cmp.Diff(beforeSnapshot, snapshotWorkingCopy, snapCmpOpts); diff != "" {
-				t.Errorf("Snapshot was modified (-initial,+end):\n%s", diff)
-			}
-		})
+				if diff := cmp.Diff(beforeSnapshot, snapshotWorkingCopy, snapCmpOpts); diff != "" {
+					t.Errorf("Snapshot was modified (-initial,+end):\n%s", diff)
+				}
+			})
+		}
 	}
 }
diff --git a/pkg/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index 5c2ac177b..8e3ee5ca2 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -4098,81 +4098,87 @@ func TestPreemption(t *testing.T) {
 		},
 	}
 	for name, tc := range cases {
-		t.Run(name, func(t *testing.T) {
-			if tc.disableLendingLimit {
-				features.SetFeatureGateDuringTest(t, features.LendingLimit, false)
-			}
-			ctx, log := utiltesting.ContextWithLog(t)
-			cl := utiltesting.NewClientBuilder().
-				WithLists(&kueue.WorkloadList{Items: tc.admitted}).
-				WithStatusSubresource(&kueue.Workload{}).
-				WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge}).
-				Build()
+		for _, useMergePatch := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s when the WorkloadRequestUseMergePatch feature is %t", name, useMergePatch), func(t *testing.T) {
+				features.SetFeatureGateDuringTest(t, features.WorkloadRequestUseMergePatch, useMergePatch)
 
-			cqCache := schdcache.New(cl)
-			for _, flv := range flavors {
-				cqCache.AddOrUpdateResourceFlavor(log, flv)
-			}
-			for _, cq := range tc.clusterQueues {
-				if err := cqCache.AddClusterQueue(ctx, cq); err != nil {
-					t.Fatalf("Couldn't add ClusterQueue to cache: %v", err)
+				if tc.disableLendingLimit {
+					features.SetFeatureGateDuringTest(t, features.LendingLimit, false)
 				}
-			}
-			for _, cohort := range tc.cohorts {
-				if err := cqCache.AddOrUpdateCohort(cohort); err != nil {
-					t.Fatalf("Couldn't add Cohort to cache: %v", err)
+				ctx, log := utiltesting.ContextWithLog(t)
+				cl := utiltesting.NewClientBuilder().
+					WithLists(&kueue.WorkloadList{Items: tc.admitted}).
+					WithStatusSubresource(&kueue.Workload{}).
+					WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge}).
+					Build()
+
+				cqCache := schdcache.New(cl)
+				for _, flv := range flavors {
+					cqCache.AddOrUpdateResourceFlavor(log, flv)
+				}
+				for _, cq := range tc.clusterQueues {
+					if err := cqCache.AddClusterQueue(ctx, cq); err != nil {
+						t.Fatalf("Couldn't add ClusterQueue to cache: %v", err)
+					}
+				}
+				for _, cohort := range tc.cohorts {
+					if err := cqCache.AddOrUpdateCohort(cohort); err != nil {
+						t.Fatalf("Couldn't add Cohort to cache: %v", err)
+					}
+				}
+
+				broadcaster := record.NewBroadcaster()
+				scheme := runtime.NewScheme()
+				if err := kueue.AddToScheme(scheme); err != nil {
+					t.Fatalf("Failed adding kueue scheme: %v", err)
 				}
-			}
+				recorder := broadcaster.NewRecorder(scheme, corev1.EventSource{Component: constants.AdmissionName})
+				preemptor := New(cl, workload.Ordering{}, recorder, nil, false, clocktesting.NewFakeClock(now))
 
-			broadcaster := record.NewBroadcaster()
-			scheme := runtime.NewScheme()
-			if err := kueue.AddToScheme(scheme); err != nil {
-				t.Fatalf("Failed adding kueue scheme: %v", err)
-			}
-			recorder := broadcaster.NewRecorder(scheme, corev1.EventSource{Component: constants.AdmissionName})
-			preemptor := New(cl, workload.Ordering{}, recorder, nil, false, clocktesting.NewFakeClock(now))
+				beforeSnapshot, err := cqCache.Snapshot(ctx)
+				if err != nil {
+					t.Fatalf("unexpected error while building snapshot: %v", err)
+				}
+				// make a working copy of the snapshotWorkingCopy than preemption can temporarily modify
+				snapshotWorkingCopy, err := cqCache.Snapshot(ctx)
+				if err != nil {
+					t.Fatalf("unexpected error while building snapshot: %v", err)
+				}
+				wlInfo := workload.NewInfo(tc.incoming)
+				wlInfo.ClusterQueue = tc.targetCQ
+				targets := preemptor.GetTargets(log, *wlInfo, tc.assignment, snapshotWorkingCopy)
+				preempted, failed, err := preemptor.IssuePreemptions(ctx, wlInfo, targets, snapshotWorkingCopy.ClusterQueue(wlInfo.ClusterQueue))
+				if err != nil {
+					t.Fatalf("Failed doing preemption")
+				}
+				if preempted != tc.wantPreempted {
+					t.Errorf("Reported %d preemptions, want %d", preempted, tc.wantPreempted)
+				}
+				if failed != 0 {
+					t.Errorf("Reported %d failed preemptions, want 0", failed)
+				}
 
-			beforeSnapshot, err := cqCache.Snapshot(ctx)
-			if err != nil {
-				t.Fatalf("unexpected error while building snapshot: %v", err)
-			}
-			// make a working copy of the snapshotWorkingCopy than preemption can temporarily modify
-			snapshotWorkingCopy, err := cqCache.Snapshot(ctx)
-			if err != nil {
-				t.Fatalf("unexpected error while building snapshot: %v", err)
-			}
-			wlInfo := workload.NewInfo(tc.incoming)
-			wlInfo.ClusterQueue = tc.targetCQ
-			targets := preemptor.GetTargets(log, *wlInfo, tc.assignment, snapshotWorkingCopy)
-			preempted, failed, err := preemptor.IssuePreemptions(ctx, wlInfo, targets, snapshotWorkingCopy.ClusterQueue(wlInfo.ClusterQueue))
-			if err != nil {
-				t.Fatalf("Failed doing preemption")
-			}
-			if preempted != tc.wantPreempted {
-				t.Errorf("Reported %d preemptions, want %d", preempted, tc.wantPreempted)
-			}
-			if failed != 0 {
-				t.Errorf("Reported %d failed preemptions, want 0", failed)
-			}
+				workloads := &kueue.WorkloadList{}
+				err = cl.List(ctx, workloads)
+				if err != nil {
+					t.Fatalf("Failed to List workloads: %v", err)
+				}
 
-			workloads := &kueue.WorkloadList{}
-			err = cl.List(ctx, workloads)
-			if err != nil {
-				t.Fatalf("Failed to List workloads: %v", err)
-			}
+				defaultCmpOpts := cmp.Options{
+					cmpopts.EquateEmpty(),
+					cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
+					cmpopts.SortSlices(func(a, b metav1.Condition) bool { return a.Type < b.Type }),
+				}
 
-			defaultCmpOpts := cmp.Options{
-				cmpopts.EquateEmpty(),
-				cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
-			}
-			if diff := cmp.Diff(tc.wantWorkloads, workloads.Items, defaultCmpOpts); diff != "" {
-				t.Errorf("Unexpected workloads (-want/+got)\n%s", diff)
-			}
+				if diff := cmp.Diff(tc.wantWorkloads, workloads.Items, defaultCmpOpts); diff != "" {
+					t.Errorf("Unexpected workloads (-want/+got)\n%s", diff)
+				}
 
-			if diff := cmp.Diff(beforeSnapshot, snapshotWorkingCopy, snapCmpOpts); diff != "" {
-				t.Errorf("Snapshot was modified (-initial,+end):\n%s", diff)
-			}
-		})
+				if diff := cmp.Diff(beforeSnapshot, snapshotWorkingCopy, snapCmpOpts); diff != "" {
+					t.Errorf("Snapshot was modified (-initial,+end):\n%s", diff)
+				}
+			})
+		}
 	}
 }
 
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 66a8cf238..5a07e3bd6 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -33,7 +33,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/client-go/tools/record"
-	"k8s.io/component-base/featuregate"
 	"k8s.io/component-base/metrics/testutil"
 	testingclock "k8s.io/utils/clock/testing"
 	"k8s.io/utils/ptr"
@@ -9125,15 +9124,11 @@ func TestSchedulerWhenWorkloadModifiedConcurrently(t *testing.T) {
 	lq := utiltestingapi.MakeLocalQueue("lq", metav1.NamespaceDefault).ClusterQueue(cq.Name).Obj()
 
 	testCases := map[string]struct {
-		features      map[featuregate.Feature]bool
 		workload      *kueue.Workload
 		wantWorkloads []kueue.Workload
 		wantEvents    []utiltesting.EventRecord
 	}{
-		"use patch in admit when the WorkloadRequestUseMergePatch feature is disabled": {
-			features: map[featuregate.Feature]bool{
-				features.WorkloadRequestUseMergePatch: false,
-			},
+		"use patch in admit": {
 			workload: utiltestingapi.MakeWorkload("wl", metav1.NamespaceDefault).
 				ResourceVersion("1").
 				Queue(kueue.LocalQueueName(lq.Name)).
@@ -9177,10 +9172,7 @@ func TestSchedulerWhenWorkloadModifiedConcurrently(t *testing.T) {
 				utiltesting.MakeEventRecord(metav1.NamespaceDefault, "wl", "Admitted", corev1.EventTypeNormal).Obj(),
 			},
 		},
-		"use patch in requeueAndUpdate when the WorkloadRequestUseMergePatch feature is disabled": {
-			features: map[featuregate.Feature]bool{
-				features.WorkloadRequestUseMergePatch: false,
-			},
+		"use patch in requeueAndUpdate": {
 			workload: utiltestingapi.MakeWorkload("wl", metav1.NamespaceDefault).
 				ResourceVersion("1").
 				Queue(kueue.LocalQueueName(lq.Name)).
@@ -9217,82 +9209,84 @@ func TestSchedulerWhenWorkloadModifiedConcurrently(t *testing.T) {
 		},
 	}
 	for name, tc := range testCases {
-		t.Run(name, func(t *testing.T) {
-			for feature, enabled := range tc.features {
-				features.SetFeatureGateDuringTest(t, feature, enabled)
-			}
-			ctx, log := utiltesting.ContextWithLog(t)
-			clientBuilder := utiltesting.NewClientBuilder().
-				WithObjects(ns.DeepCopy(), rf.DeepCopy(), cq.DeepCopy(), lq.DeepCopy(), tc.workload).
-				WithStatusSubresource(&kueue.Workload{}).
-				WithInterceptorFuncs(interceptor.Funcs{
-					SubResourcePatch: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
-						if _, ok := obj.(*kueue.Workload); ok && subResourceName == "status" {
-							// Simulate concurrent modification by another controller
-							wlCopy := tc.workload.DeepCopy()
-							if wlCopy.Labels == nil {
-								wlCopy.Labels = make(map[string]string, 1)
-							}
-							wlCopy.Labels["test.kueue.x-k8s.io/timestamp"] = time.Now().String()
-							if err := c.Update(ctx, wlCopy); err != nil {
-								return err
+		for _, enabled := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s when the WorkloadRequestUseMergePatch feature is %t", name, enabled), func(t *testing.T) {
+				features.SetFeatureGateDuringTest(t, features.WorkloadRequestUseMergePatch, enabled)
+				ctx, log := utiltesting.ContextWithLog(t)
+				var patched bool
+				clientBuilder := utiltesting.NewClientBuilder().
+					WithObjects(ns.DeepCopy(), rf.DeepCopy(), cq.DeepCopy(), lq.DeepCopy(), tc.workload).
+					WithStatusSubresource(&kueue.Workload{}).
+					WithInterceptorFuncs(interceptor.Funcs{
+						SubResourcePatch: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+							if _, ok := obj.(*kueue.Workload); ok && subResourceName == "status" && !patched {
+								patched = true
+								// Simulate concurrent modification by another controller
+								wlCopy := tc.workload.DeepCopy()
+								if wlCopy.Labels == nil {
+									wlCopy.Labels = make(map[string]string, 1)
+								}
+								wlCopy.Labels["test.kueue.x-k8s.io/timestamp"] = time.Now().String()
+								if err := c.Update(ctx, wlCopy); err != nil {
+									return err
+								}
 							}
-						}
-						return utiltesting.TreatSSAAsStrategicMerge(ctx, c, subResourceName, obj, patch, opts...)
-					},
-				})
-			cl := clientBuilder.Build()
-			recorder := &utiltesting.EventRecorder{}
+							return utiltesting.TreatSSAAsStrategicMerge(ctx, c, subResourceName, obj, patch, opts...)
+						},
+					})
+				cl := clientBuilder.Build()
+				recorder := &utiltesting.EventRecorder{}
 
-			cqCache := schdcache.New(cl)
-			qManager := qcache.NewManager(cl, cqCache)
+				cqCache := schdcache.New(cl)
+				qManager := qcache.NewManager(cl, cqCache)
 
-			cqCache.AddOrUpdateResourceFlavor(log, rf.DeepCopy())
-			if err := cqCache.AddClusterQueue(ctx, cq.DeepCopy()); err != nil {
-				t.Fatalf("Inserting clusterQueue %s in cache: %v", cq.Name, err)
-			}
-			if err := qManager.AddClusterQueue(ctx, cq.DeepCopy()); err != nil {
-				t.Fatalf("Inserting clusterQueue %s in manager: %v", cq.Name, err)
-			}
-			if err := qManager.AddLocalQueue(ctx, lq.DeepCopy()); err != nil {
-				t.Fatalf("Inserting queue %s/%s in manager: %v", lq.Namespace, lq.Name, err)
-			}
+				cqCache.AddOrUpdateResourceFlavor(log, rf.DeepCopy())
+				if err := cqCache.AddClusterQueue(ctx, cq.DeepCopy()); err != nil {
+					t.Fatalf("Inserting clusterQueue %s in cache: %v", cq.Name, err)
+				}
+				if err := qManager.AddClusterQueue(ctx, cq.DeepCopy()); err != nil {
+					t.Fatalf("Inserting clusterQueue %s in manager: %v", cq.Name, err)
+				}
+				if err := qManager.AddLocalQueue(ctx, lq.DeepCopy()); err != nil {
+					t.Fatalf("Inserting queue %s/%s in manager: %v", lq.Namespace, lq.Name, err)
+				}
 
-			scheduler := New(qManager, cqCache, cl, recorder, WithClock(t, testingclock.NewFakeClock(now)))
+				scheduler := New(qManager, cqCache, cl, recorder, WithClock(t, testingclock.NewFakeClock(now)))
 
-			wg := sync.WaitGroup{}
-			scheduler.setAdmissionRoutineWrapper(routine.NewWrapper(
-				func() { wg.Add(1) },
-				func() { wg.Done() },
-			))
+				wg := sync.WaitGroup{}
+				scheduler.setAdmissionRoutineWrapper(routine.NewWrapper(
+					func() { wg.Add(1) },
+					func() { wg.Done() },
+				))
 
-			ctx, cancel := context.WithTimeout(ctx, queueingTimeout)
-			go qManager.CleanUpOnContext(ctx)
-			defer cancel()
+				ctx, cancel := context.WithTimeout(ctx, queueingTimeout)
+				go qManager.CleanUpOnContext(ctx)
+				defer cancel()
 
-			scheduler.schedule(ctx)
-			wg.Wait()
+				scheduler.schedule(ctx)
+				wg.Wait()
 
-			gotWorkloads := &kueue.WorkloadList{}
-			err := cl.List(ctx, gotWorkloads)
-			if err != nil {
-				t.Fatalf("Unexpected list workloads error: %v", err)
-			}
+				gotWorkloads := &kueue.WorkloadList{}
+				err := cl.List(ctx, gotWorkloads)
+				if err != nil {
+					t.Fatalf("Unexpected list workloads error: %v", err)
+				}
 
-			opts := cmp.Options{
-				cmpopts.EquateEmpty(),
-				cmpopts.IgnoreFields(metav1.ObjectMeta{}, "Labels"),
-			}
+				opts := cmp.Options{
+					cmpopts.EquateEmpty(),
+					cmpopts.IgnoreFields(metav1.ObjectMeta{}, "Labels"),
+				}
 
-			if diff := cmp.Diff(tc.wantWorkloads, gotWorkloads.Items, opts); diff != "" {
-				t.Errorf("Unexpected scheduled workloads (-want,+got):\n%s", diff)
-			}
+				if diff := cmp.Diff(tc.wantWorkloads, gotWorkloads.Items, opts); diff != "" {
+					t.Errorf("Unexpected scheduled workloads (-want,+got):\n%s", diff)
+				}
 
-			if diff := cmp.Diff(tc.wantEvents, recorder.RecordedEvents,
-				cmpopts.IgnoreFields(utiltesting.EventRecord{}, "Message"),
-			); diff != "" {
-				t.Errorf("Unexpected events (-want/+got):\n%s", diff)
-			}
-		})
+				if diff := cmp.Diff(tc.wantEvents, recorder.RecordedEvents,
+					cmpopts.IgnoreFields(utiltesting.EventRecord{}, "Message"),
+				); diff != "" {
+					t.Errorf("Unexpected events (-want/+got):\n%s", diff)
+				}
+			})
+		}
 	}
 }
diff --git a/pkg/util/client/client_test.go b/pkg/util/client/client_test.go
index 83555a55d..5291a7b3b 100644
--- a/pkg/util/client/client_test.go
+++ b/pkg/util/client/client_test.go
@@ -17,16 +17,19 @@ limitations under the License.
 package client
 
 import (
+	"context"
 	"errors"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	batchv1 "k8s.io/api/batch/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 )
@@ -72,8 +75,9 @@ func TestPatch(t *testing.T) {
 		options []PatchOption
 	}
 	type want struct {
-		err error
-		obj client.Object // To assert patched object.
+		fetched bool
+		err     error
+		obj     client.Object // To assert patched object.
 	}
 	// clientObject is used to initialize test Client in t.Run().
 	clientObject := newObject("2")
@@ -93,8 +97,27 @@ func TestPatch(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errTestConflict,
-				obj: newObject("2"),
+				fetched: false,
+				err:     errTestConflict,
+				obj:     newObject("2"),
+			},
+		},
+		"Strict_OutdatedLocalObject_RetryOnConflict": {
+			args: args{
+				obj: newObject("1"), // outdated local object.
+				update: func(job *batchv1.Job) UpdateFunc {
+					return func() (bool, error) {
+						job.Spec.Suspend = ptr.To(true)
+						return true, nil
+					}
+				},
+				options: []PatchOption{WithRetryOnConflict()},
+			},
+			want: want{
+				fetched: true,
+				obj: newObject("3", func(job *batchv1.Job) {
+					job.Spec.Suspend = ptr.To(true)
+				}),
 			},
 		},
 		"Strict_CurrentLocalObject": {
@@ -115,6 +138,37 @@ func TestPatch(t *testing.T) {
 				}),
 			},
 		},
+		"Strict_ConflictError": {
+			args: args{
+				obj: newObject("2"),
+				update: func(job *batchv1.Job) UpdateFunc {
+					return func() (bool, error) {
+						return false, errTestConflict
+					}
+				},
+			},
+			want: want{
+				fetched: false,
+				err:     errTestConflict,
+				obj:     newObject("2"),
+			},
+		},
+		"Strict_ConflictError_RetryOnConflict": {
+			args: args{
+				obj: newObject("2"),
+				update: func(job *batchv1.Job) UpdateFunc {
+					return func() (bool, error) {
+						return false, errTestConflict
+					}
+				},
+				options: []PatchOption{WithRetryOnConflict()},
+			},
+			want: want{
+				fetched: true,
+				err:     errTestConflict,
+				obj:     newObject("2"),
+			},
+		},
 		"NotStrict_OutdatedLocalObject": {
 			args: args{
 				obj: newObject("1"), // outdated local object.
@@ -180,11 +234,29 @@ func TestPatch(t *testing.T) {
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
 			ctx, _ := utiltesting.ContextWithLog(t)
-			clnt := utiltesting.NewClientBuilder().WithObjects(clientObject).Build()
+			fetched := false
+			clnt := utiltesting.NewClientBuilder().WithObjects(clientObject).
+				WithInterceptorFuncs(interceptor.Funcs{
+					Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+						fetched = true
+						return c.Get(ctx, key, obj, opts...)
+					},
+					Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error {
+						return c.Update(ctx, obj, opts...)
+					},
+				}).
+				Build()
 			err := Patch(ctx, clnt, tt.args.obj, tt.args.update(tt.args.obj), tt.args.options...)
-			if diff := cmp.Diff(tt.want.err, err); diff != "" {
+			var errOpts cmp.Options
+			if !apierrors.IsConflict(err) {
+				errOpts = append(errOpts, cmpopts.EquateErrors())
+			}
+			if diff := cmp.Diff(tt.want.err, err, errOpts...); diff != "" {
 				t.Errorf("unexpected error (-want/+got):\n%s", diff)
 			}
+			if diff := cmp.Diff(tt.want.fetched, fetched); diff != "" {
+				t.Errorf("unexpected fetched (-want/+got):\n%s", diff)
+			}
 			if err := clnt.Get(ctx, client.ObjectKeyFromObject(tt.args.obj), tt.args.obj); err != nil {
 				t.Fatalf("Patch() unexpected error getting object: %v", err)
 			}
@@ -202,8 +274,9 @@ func TestPatchStatus(t *testing.T) {
 		options []PatchOption
 	}
 	type want struct {
-		err error
-		obj client.Object // To assert patched object.
+		fetched bool
+		err     error
+		obj     client.Object // To assert patched object.
 	}
 	// clientObject is used to initialize test Client in t.Run().
 	clientObject := newObject("2")
@@ -223,8 +296,27 @@ func TestPatchStatus(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errTestConflict,
-				obj: newObject("2"),
+				fetched: false,
+				err:     errTestConflict,
+				obj:     newObject("2"),
+			},
+		},
+		"Strict_OutdatedLocalObject_RetryOnConflict": {
+			args: args{
+				obj: newObject("1"), // outdated local object results in patch error.
+				update: func(job *batchv1.Job) UpdateFunc {
+					return func() (bool, error) {
+						job.Status.Active = 1
+						return true, nil
+					}
+				},
+				options: []PatchOption{WithRetryOnConflict()},
+			},
+			want: want{
+				fetched: true,
+				obj: newObject("3", func(job *batchv1.Job) {
+					job.Status.Active = 1
+				}),
 			},
 		},
 		"Strict_CurrentLocalObject": {
@@ -245,6 +337,37 @@ func TestPatchStatus(t *testing.T) {
 				}),
 			},
 		},
+		"Strict_ConflictError": {
+			args: args{
+				obj: newObject("2"),
+				update: func(job *batchv1.Job) UpdateFunc {
+					return func() (bool, error) {
+						return false, errTestConflict
+					}
+				},
+			},
+			want: want{
+				fetched: false,
+				err:     errTestConflict,
+				obj:     newObject("2"),
+			},
+		},
+		"Strict_ConflictError_RetryOnConflict": {
+			args: args{
+				obj: newObject("2"),
+				update: func(job *batchv1.Job) UpdateFunc {
+					return func() (bool, error) {
+						return false, errTestConflict
+					}
+				},
+				options: []PatchOption{WithRetryOnConflict()},
+			},
+			want: want{
+				fetched: true,
+				err:     errTestConflict,
+				obj:     newObject("2"),
+			},
+		},
 		"NotStrict_OutdatedLocalObject": {
 			args: args{
 				obj: newObject("1"), // outdated local object.
@@ -311,11 +434,29 @@ func TestPatchStatus(t *testing.T) {
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
 			ctx, _ := utiltesting.ContextWithLog(t)
-			clnt := utiltesting.NewClientBuilder().WithObjects(clientObject).Build()
+			fetched := false
+			clnt := utiltesting.NewClientBuilder().WithObjects(clientObject).
+				WithInterceptorFuncs(interceptor.Funcs{
+					Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+						fetched = true
+						return c.Get(ctx, key, obj, opts...)
+					},
+					SubResourcePatch: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+						return c.Status().Patch(ctx, obj, patch, opts...)
+					},
+				}).
+				Build()
 			err := PatchStatus(ctx, clnt, tt.args.obj, tt.args.update(tt.args.obj), tt.args.options...)
-			if diff := cmp.Diff(tt.want.err, err); diff != "" {
+			var errOpts cmp.Options
+			if !apierrors.IsConflict(err) {
+				errOpts = append(errOpts, cmpopts.EquateErrors())
+			}
+			if diff := cmp.Diff(tt.want.err, err, errOpts...); diff != "" {
 				t.Errorf("unexpected error (-want/+got):\n%s", diff)
 			}
+			if diff := cmp.Diff(tt.want.fetched, fetched); diff != "" {
+				t.Errorf("unexpected fetched (-want/+got):\n%s", diff)
+			}
 			if err := clnt.Get(ctx, client.ObjectKeyFromObject(tt.args.obj), tt.args.obj); err != nil {
 				t.Fatalf("Patch() unexpected error getting object: %v", err)
 			}
diff --git a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
index bc40df123..ab419a342 100644
--- a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
+++ b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/admissionchecks/provisioning"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	utiltestingapi "sigs.k8s.io/kueue/pkg/util/testing/v1beta2"
@@ -1091,14 +1092,15 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 			ginkgo.By("Checking if the workload is deactivated, once had Rejected status in the status.admissionCheck[*] field, an event is emitted and a metric is increased", func() {
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
-
-					g.Expect(updatedWl.Status.AdmissionChecks).To(gomega.ContainElement(gomega.BeComparableTo(
-						kueue.AdmissionCheckState{
-							Name:       kueue.AdmissionCheckReference(ac.Name),
-							State:      kueue.CheckStatePending,
-							Message:    "Reset to Pending after eviction. Previously: Rejected",
-							RetryCount: ptr.To(int32(1)),
-						},
+					wantState := kueue.AdmissionCheckState{
+						Name:    kueue.AdmissionCheckReference(ac.Name),
+						State:   kueue.CheckStatePending,
+						Message: "Reset to Pending after eviction. Previously: Rejected",
+					}
+					if !features.Enabled(features.WorkloadRequestUseMergePatch) {
+						wantState.RetryCount = ptr.To(int32(1))
+					}
+					g.Expect(updatedWl.Status.AdmissionChecks).To(gomega.ContainElement(gomega.BeComparableTo(wantState,
 						cmpopts.IgnoreFields(kueue.AdmissionCheckState{}, "LastTransitionTime", "PodSetUpdates", "RequeueAfterSeconds"))))
 					g.Expect(workload.IsActive(&updatedWl)).To(gomega.BeFalse())