Repository: kubernetes-sigs/kueue

Issue:
Kubeflow Trainer v2 integration follow ups
<!-- Please only use this template for submitting clean up requests -->

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

This is to address remaining comments to https://github.com/kubernetes-sigs/kueue/pull/6597

1. Don't lookup the JobSet object in the GenericJob implementation
2. Don't use the Background context, instead use the "request" context from PodSets() etc.


**Why is this needed**:
1 because:
a) JobSet should be an implementation detail of the Trainer which might be replaced in the future
b) it creates tricky race conditions to reason about, because the JobSet may not be immediately created
c) it couples the JobSet API version used for Trainer with the version we integrate with

Instead, helpers exposed by the Trainer should be used, but they require releasing 2.0.1.

2. becuase:
1. background context cannot be cancelled, so the goroutine leaks / remains running even if the main Kueue goroutine is terminated
2. the "request" context may have more data to log, like the workload name etc.

Discussion:
@mimowo: /assign @kaisoz 
Who I think is already looking into the follow ups.
cc @tenzen-y @andreyvelich @astefanutti @kannon92: Also can we add a page in our site for trainer?

cc @andreyvelich @kaisoz@mimowo: For sure we need docs, but we can add docs post code release, potentially. 

So, I would prefer to focus on addressing the code follow ups. @izturn: Just tried the current code and it doesn’t seem to work in **TAS** mode. Looks like the Pods don’t get the proper podset/workload labels and annotations when they start up(at `RunWithPodSetsInfo`). Because of that, the ungater can’t find the matching Pods(at `podsForPodSet`), so the scheduling gate never gets removed and the pods just stays stuck in **Pending**. @kaisoz @mimowo: Yes, this is the current "known limitation" due to the fact that Trainer v2 does not allow to update the podSetOverrides with labels & annotations, see open issue: https://github.com/kubernetes-sigs/kueue/issues/6865

It would be great to work on resoling this limitation.

cc @mwysokin @mimowo: Actually, there might be a dirty workaround inside Kueue. We could access the JobSet object and inject the labels & annotations before unsuspending in the Resume phase. 

If we have a PR, I could support it as a "temporary fix" while waiting for Trainer v2 fixes.@mimowo: cc @kaisoz @andreyvelich @tenzen-y @kaisoz: > Actually, there might be a dirty workaround inside Kueue. We could access the JobSet object and inject the labels & annotations before unsuspending in the Resume phase. 
> 
> If we have a PR, I could support it as a "temporary fix" while waiting for Trainer v2 fixes.

Unfortunately that won't work @mimowo because the trainjob controller reconciles the jobset and removes the labels and annotations (I've tried this already)@mimowo: Got it, what about mutating the Trainer's Runtime object before resuming@andreyvelich: @mimowo @kaisoz Can we configure those labels/annotations in the Runtime before TrainJob is created?
Or labels must be set by the Kueue controller ?@izturn: the `podsForPodSet`'s filter is the podset & workload's name, so i guess they can't be configured?

> [@mimowo](https://github.com/mimowo) [@kaisoz](https://github.com/kaisoz) Can we configure those labels/annotations in the Runtime before TrainJob is created? Or labels must be set by the Kueue controller ?

@mimowo: > Can we configure those labels/annotations in the Runtime before TrainJob is created?

before created I don't think, because Kueue may admit / evict multiple times during lifetime.

We could however "modify" the TainerRuntime object, unless it is explicitly prohibited by a webhook?@andreyvelich: >before created I don't think, because Kueue may admit / evict multiple times during lifetime.

Does Kueue modify the TAS labels during admission or eviction ? @mimowo: > Does Kueue modify the TAS labels during admission or eviction ?

Both@mimowo: @izturn one more thought - can you check using "main" Kueue. I suspect that maybe there is a difference after we stopped injecting the "tas" label. Still, we will not support the "implicit" mode, but I suspect that "required" and "preferred" modes have a chance to work.@tenzen-y: Additionally, I would like to narrow down the below annotation usage by renaming to `kueue.x-k8s.io/trainjob-override-idx` so that we can avoid accidentally introduce the annotation to different places.

https://github.com/kubernetes-sigs/kueue/blob/4c66b01c67e972c7250eeeb8ffeb87ccf961054a/pkg/controller/jobs/kubeflow/trainjob/trainjob_controller.go#L53
@mimowo: Ultimately I think the podSetOverrides should have the 'manager' field, then the annotation would not be needed. I discussed this with  Tomas and we came up with the annotation idea as a workaround, so that Kueue can know which overrides are done by Kueue and which by user.@kaisoz: > Got it, what about mutating the Trainer's Runtime object before resuming

I think that should work! Although it'd make both the `RunWithPodsetsInfo` and `RestorePodSetsInfo` a bit more complicated. Do you want me to try that out or better wait for Trainer PR to sink in?@kaisoz: > Additionally, I would like to narrow down the below annotation usage by renaming to `kueue.x-k8s.io/trainjob-override-idx` so that we can avoid accidentally introduce the annotation to different places.
> 
> [kueue/pkg/controller/jobs/kubeflow/trainjob/trainjob_controller.go](https://github.com/kubernetes-sigs/kueue/blob/4c66b01c67e972c7250eeeb8ffeb87ccf961054a/pkg/controller/jobs/kubeflow/trainjob/trainjob_controller.go#L53)
> 
> Line 53 in [4c66b01](/kubernetes-sigs/kueue/commit/4c66b01c67e972c7250eeeb8ffeb87ccf961054a)
> 
>  FirstOverrideIdx = "kueue.x-k8s.io/override-idx"

As @mimowo said, this is a workaround,but anyway I can rename the annotation 👍🏻 @izturn: > [@izturn](https://github.com/izturn) one more thought - can you check using "main" Kueue. I suspect that maybe there is a difference after we stopped injecting the "tas" label. Still, we will not support the "implicit" mode, but I suspect that "required" and "preferred" modes have a chance to work.

Sure, I can try again. But tbh, as a beginner, I don’t really understand what you mean by the “main” kueue things. I’m not sure how to do that. Could you walk me through it with a bit more detail? Thanks!@tenzen-y: > Ultimately I think the podSetOverrides should have the 'manager' field, then the annotation would not be needed. I discussed this with Tomas and we came up with the annotation idea as a workaround, so that Kueue can know which overrides are done by Kueue and which by user.

We probably need to generize the case for manager, then propose it to KF Trainer and it will not contain in the next Kueue release (v0.14).
Meanwhile, I would like to use `kueue.x-k8s.io/trainjob-override-idx` or `alpha.kueue.x-k8s.io/trainjob-override-idx` annotation.
I'm leaning towards annotation with "alpha" prefix since it should be a temporary annotation.
@mimowo: Regarding the alpha prefix I think there used to be this guidance in the past, but the API reviewers moved away from putting maturity level into the label or annotation name. I added the backOffLinitPerIndex feature which introduces annotation and I wasn't asked to add the prefix. Also I see other Alpha annotations in 1.34 not adding this https://kubernetes.io/docs/reference/labels-annotations-taints/?utm_source=chatgpt.com#applyset-kubernetes-io-additional-namespaces
so the maturity level is emphasized in the docs, but not in the annotation name.

I think if we just add to comment: This is alpha level annotation it should be enough. We could also call it deprecated already which maybe is better maturity level. I will open an issue in trainer.@mimowo: As part of this cleanup task we can unexport the const field @tenzen-y: > Regarding the alpha prefix I think there used to be this guidance in the past, but the API reviewers moved away from putting maturity level into the label or annotation name. I added the backOffLinitPerIndex feature which introduces annotation and I wasn't asked to add the prefix. Also I see other Alpha annotations in 1.34 not adding this https://kubernetes.io/docs/reference/labels-annotations-taints/?utm_source=chatgpt.com#applyset-kubernetes-io-additional-namespaces so the maturity level is emphasized in the docs, but not in the annotation name.
> 
> I think if we just add to comment: This is alpha level annotation it should be enough. We could also call it deprecated already which maybe is better maturity level. I will open an issue in trainer.

That makes sense. In that case, let's use `kueue.x-k8s.io/trainjob-override-idx`.@mimowo: > > [@izturn](https://github.com/izturn) one more thought - can you check using "main" Kueue. I suspect that maybe there is a difference after we stopped injecting the "tas" label. Still, we will not support the "implicit" mode, but I suspect that "required" and "preferred" modes have a chance to work.
> 
> Sure, I can try again. But tbh, as a beginner, I don’t really understand what you mean by the “main” kueue things. I’m not sure how to do that. Could you walk me through it with a bit more detail? Thanks!


Sure, by "main" I meant the latest build of Kueue from the main branch. You can install for example following https://kueue.sigs.k8s.io/docs/installation/#install-the-latest-development-version


However, I think it will not work anyway on the second thought, because we would not be able to inject the "workload" annotation :/ @izturn: > > > [@izturn](https://github.com/izturn) one more thought - can you check using "main" Kueue. I suspect that maybe there is a difference after we stopped injecting the "tas" label. Still, we will not support the "implicit" mode, but I suspect that "required" and "preferred" modes have a chance to work.
> > 
> > 
> > Sure, I can try again. But tbh, as a beginner, I don’t really understand what you mean by the “main” kueue things. I’m not sure how to do that. Could you walk me through it with a bit more detail? Thanks!
> 
> Sure, by "main" I meant the latest build of Kueue from the main branch. You can install for example following https://kueue.sigs.k8s.io/docs/installation/#install-the-latest-development-version
> 
> However, I think it will not work anyway on the second thought, because we would not be able to inject the "workload" annotation :/

Oh, got it. My reply above was actually based on testing with the main branch.@kaisoz: /reopen 

closed by mistake, `2` should still be addressed@k8s-ci-robot: @kaisoz: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/6865#issuecomment-3350610114):

>/reopen 
>
>closed by mistake, `2` should still be addressed


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@tenzen-y: @mimowo @kaisoz Additionally, I would request 2 things as I mentioned in https://github.com/kubernetes-sigs/kueue/pull/7030#pullrequestreview-3283582303.
@kaisoz: > [@mimowo](https://github.com/mimowo) [@kaisoz](https://github.com/kaisoz) Additionally, I would request 2 things as I mentioned in [#7030 (review)](https://github.com/kubernetes-sigs/kueue/pull/7030#pullrequestreview-3283582303).

Hi @tenzen-y , just saw this, I'll take care of it 😊 
@tenzen-y: > > [@mimowo](https://github.com/mimowo) [@kaisoz](https://github.com/kaisoz) Additionally, I would request 2 things as I mentioned in [#7030 (review)](https://github.com/kubernetes-sigs/kueue/pull/7030#pullrequestreview-3283582303).
> 
> Hi [@tenzen-y](https://github.com/tenzen-y) , just saw this, I'll take care of it 😊

Thank you!@tenzen-y: x-ref: https://github.com/kubernetes-sigs/kueue/pull/7081#discussion_r2391731893@kaisoz: > x-ref: [#7081 (comment)](https://github.com/kubernetes-sigs/kueue/pull/7081#discussion_r2391731893)

PR ready @tenzen-y  https://github.com/kubernetes-sigs/kueue/pull/7110
Thanks!

Solution patch:
diff --git a/cmd/importer/pod/import.go b/cmd/importer/pod/import.go
index 68403d36b..5b52381d4 100644
--- a/cmd/importer/pod/import.go
+++ b/cmd/importer/pod/import.go
@@ -201,7 +201,9 @@ func createWorkload(ctx context.Context, c client.Client, wl *kueue.Workload) er
 
 func admitWorkload(ctx context.Context, c client.Client, wl *kueue.Workload) error {
 	var realClock = clock.RealClock{}
-	err := workload.ApplyAdmissionStatus(ctx, c, wl, false, realClock)
+	err := workload.PatchAdmissionStatus(ctx, c, wl, true, realClock, func() (*kueue.Workload, bool, error) {
+		return wl, true, nil
+	})
 	retry, _, timeout := checkError(err)
 	for retry {
 		if timeout >= 0 {
@@ -211,7 +213,9 @@ func admitWorkload(ctx context.Context, c client.Client, wl *kueue.Workload) err
 			case <-time.After(timeout):
 			}
 		}
-		err = workload.ApplyAdmissionStatus(ctx, c, wl, false, realClock)
+		err = workload.PatchAdmissionStatus(ctx, c, wl, true, realClock, func() (*kueue.Workload, bool, error) {
+			return wl, true, nil
+		})
 		retry, _, timeout = checkError(err)
 	}
 	return err
diff --git a/pkg/controller/admissionchecks/multikueue/workload.go b/pkg/controller/admissionchecks/multikueue/workload.go
index 989523569..52ded1ce6 100644
--- a/pkg/controller/admissionchecks/multikueue/workload.go
+++ b/pkg/controller/admissionchecks/multikueue/workload.go
@@ -383,22 +383,23 @@ func (w *wlReconciler) reconcileGroup(ctx context.Context, group *wlGroup) (reco
 		}
 
 		if acs.State != kueue.CheckStateRetry && acs.State != kueue.CheckStateRejected {
-			if group.jobAdapter.KeepAdmissionCheckPending() {
-				acs.State = kueue.CheckStatePending
-			} else {
-				acs.State = kueue.CheckStateReady
-			}
-			// update the message
-			acs.Message = fmt.Sprintf("The workload got reservation on %q", reservingRemote)
-			// update the transition time since is used to detect the lost worker state.
-			acs.LastTransitionTime = metav1.NewTime(w.clock.Now())
-
-			workload.SetAdmissionCheckState(&group.local.Status.AdmissionChecks, *acs, w.clock)
-			// Set the cluster name to the reserving remote and clear the nominated clusters.
-			group.local.Status.ClusterName = &reservingRemote
-			group.local.Status.NominatedClusterNames = nil
-
-			if err := workload.ApplyAdmissionStatus(ctx, w.client, group.local, true, w.clock); err != nil {
+			if err := workload.PatchAdmissionStatus(ctx, w.client, group.local, true, w.clock, func() (*kueue.Workload, bool, error) {
+				if group.jobAdapter.KeepAdmissionCheckPending() {
+					acs.State = kueue.CheckStatePending
+				} else {
+					acs.State = kueue.CheckStateReady
+				}
+				// update the message
+				acs.Message = fmt.Sprintf("The workload got reservation on %q", reservingRemote)
+				// update the transition time since is used to detect the lost worker state.
+				acs.LastTransitionTime = metav1.NewTime(w.clock.Now())
+
+				workload.SetAdmissionCheckState(&group.local.Status.AdmissionChecks, *acs, w.clock)
+				// Set the cluster name to the reserving remote and clear the nominated clusters.
+				group.local.Status.ClusterName = &reservingRemote
+				group.local.Status.NominatedClusterNames = nil
+				return group.local, true, nil
+			}); err != nil {
 				log.V(2).Error(err, "Failed to patch workload", "workload", klog.KObj(group.local))
 				return reconcile.Result{}, err
 			}
@@ -435,8 +436,10 @@ func (w *wlReconciler) nominateAndSynchronizeWorkers(ctx context.Context, group
 			nominatedWorkers = append(nominatedWorkers, workerName)
 		}
 		if group.local.Status.ClusterName == nil && !equality.Semantic.DeepEqual(group.local.Status.NominatedClusterNames, nominatedWorkers) {
-			group.local.Status.NominatedClusterNames = nominatedWorkers
-			if err := workload.ApplyAdmissionStatus(ctx, w.client, group.local, true, w.clock); err != nil {
+			if err := workload.PatchAdmissionStatus(ctx, w.client, group.local, true, w.clock, func() (*kueue.Workload, bool, error) {
+				group.local.Status.NominatedClusterNames = nominatedWorkers
+				return group.local, true, nil
+			}); err != nil {
 				log.V(2).Error(err, "Failed to patch nominated clusters", "workload", klog.KObj(group.local))
 				return reconcile.Result{}, err
 			}
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 3be9a9529..0e7dfbb5d 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -195,10 +195,10 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		r.workloadHasDRA(&wl) {
 		if r.workloadHasResourceClaim(&wl) {
 			log.V(3).Info("Workload is inadmissible because it uses resource claims which is not supported")
-			if workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, "DynamicResourceAllocation feature does not support use of resource claims", r.clock.Now()) {
-				if err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock); err != nil {
-					return ctrl.Result{}, fmt.Errorf("failed to update workload status for DRA resource claims error: %w", err)
-				}
+			if err := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+				return &wl, workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, "DynamicResourceAllocation feature does not support use of resource claims", r.clock.Now()), nil
+			}); err != nil {
+				return ctrl.Result{}, fmt.Errorf("failed to update workload status for DRA resource claims error: %w", err)
 			}
 			return ctrl.Result{}, nil
 		}
@@ -207,10 +207,10 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		draResources, err := dra.GetResourceRequestsForResourceClaimTemplates(ctx, r.client, &wl)
 		if err != nil {
 			log.Error(err, "Failed to process DRA resources for workload")
-			if workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, err.Error(), r.clock.Now()) {
-				if updateErr := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock); updateErr != nil {
-					return ctrl.Result{}, fmt.Errorf("failed to update workload status for DRA error: %w", updateErr)
-				}
+			if updateErr := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+				return &wl, workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, err.Error(), r.clock.Now()), nil
+			}); updateErr != nil {
+				return ctrl.Result{}, fmt.Errorf("failed to update workload status for DRA error: %w", updateErr)
 			}
 			return ctrl.Result{}, err
 		}
@@ -241,29 +241,36 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		}
 
 		var updated bool
-		if cond := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadRequeued); cond != nil && cond.Status == metav1.ConditionFalse {
-			switch cond.Reason {
-			case kueue.WorkloadDeactivated:
-				workload.SetRequeuedCondition(&wl, kueue.WorkloadReactivated, "The workload was reactivated", true)
-				updated = true
-			case kueue.WorkloadEvictedByPodsReadyTimeout, kueue.WorkloadEvictedByAdmissionCheck:
-				var requeueAfter time.Duration
-				if wl.Status.RequeueState != nil && wl.Status.RequeueState.RequeueAt != nil {
-					requeueAfter = wl.Status.RequeueState.RequeueAt.Sub(r.clock.Now())
-				}
-				if requeueAfter > 0 {
-					return reconcile.Result{RequeueAfter: requeueAfter}, nil
-				}
-				if wl.Status.RequeueState != nil {
-					wl.Status.RequeueState.RequeueAt = nil
+		var requeueAfter time.Duration
+		err := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			if cond := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadRequeued); cond != nil && cond.Status == metav1.ConditionFalse {
+				switch cond.Reason {
+				case kueue.WorkloadDeactivated:
+					workload.SetRequeuedCondition(&wl, kueue.WorkloadReactivated, "The workload was reactivated", true)
+					updated = true
+				case kueue.WorkloadEvictedByPodsReadyTimeout, kueue.WorkloadEvictedByAdmissionCheck:
+
+					if wl.Status.RequeueState != nil && wl.Status.RequeueState.RequeueAt != nil {
+						requeueAfter = wl.Status.RequeueState.RequeueAt.Sub(r.clock.Now())
+					}
+					if requeueAfter > 0 {
+						return nil, updated, nil
+					}
+					if wl.Status.RequeueState != nil {
+						wl.Status.RequeueState.RequeueAt = nil
+					}
+					workload.SetRequeuedCondition(&wl, kueue.WorkloadBackoffFinished, "The workload backoff was finished", true)
+					updated = true
 				}
-				workload.SetRequeuedCondition(&wl, kueue.WorkloadBackoffFinished, "The workload backoff was finished", true)
-				updated = true
 			}
-		}
+			return &wl, updated, nil
+		})
 
+		if requeueAfter > 0 {
+			return reconcile.Result{RequeueAfter: requeueAfter}, nil
+		}
 		if updated {
-			return ctrl.Result{}, client.IgnoreNotFound(workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock))
+			return ctrl.Result{}, client.IgnoreNotFound(err)
 		}
 	} else {
 		var updated, evicted bool
@@ -279,6 +286,7 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			updated = true
 			evicted = true
 		}
+		wlOrig := wl.DeepCopy()
 		if dtCond != nil {
 			apimeta.RemoveStatusCondition(&wl.Status.Conditions, kueue.WorkloadDeactivationTarget)
 		}
@@ -294,7 +302,9 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 					}
 				}
 			} else {
-				if err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock); err != nil {
+				if err := workload.PatchAdmissionStatus(ctx, r.client, wlOrig, true, r.clock, func() (*kueue.Workload, bool, error) {
+					return &wl, true, nil
+				}); err != nil {
 					if !apierrors.IsNotFound(err) {
 						return ctrl.Result{}, fmt.Errorf("updating workload: %w", err)
 					}
@@ -313,8 +323,10 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	lqExists := err == nil
 	lqActive := ptr.Deref(lq.Spec.StopPolicy, kueue.None) == kueue.None
 	if lqExists && lqActive && isDisabledRequeuedByLocalQueueStopped(&wl) {
-		workload.SetRequeuedCondition(&wl, kueue.WorkloadLocalQueueRestarted, "The LocalQueue was restarted after being stopped", true)
-		return ctrl.Result{}, client.IgnoreNotFound(workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock))
+		return ctrl.Result{}, client.IgnoreNotFound(workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			workload.SetRequeuedCondition(&wl, kueue.WorkloadLocalQueueRestarted, "The LocalQueue was restarted after being stopped", true)
+			return &wl, true, nil
+		}))
 	}
 
 	cqName, cqOk := r.queues.ClusterQueueForWorkload(&wl)
@@ -326,8 +338,10 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		}
 		// If stopped cluster queue is started we need to set the WorkloadRequeued condition to true.
 		if isDisabledRequeuedByClusterQueueStopped(&wl) && ptr.Deref(cq.Spec.StopPolicy, kueue.None) == kueue.None {
-			workload.SetRequeuedCondition(&wl, kueue.WorkloadClusterQueueRestarted, "The ClusterQueue was restarted after being stopped", true)
-			return ctrl.Result{}, client.IgnoreNotFound(workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock))
+			return ctrl.Result{}, client.IgnoreNotFound(workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+				workload.SetRequeuedCondition(&wl, kueue.WorkloadClusterQueueRestarted, "The ClusterQueue was restarted after being stopped", true)
+				return &wl, true, nil
+			}))
 		}
 		if updated, err := r.reconcileSyncAdmissionChecks(ctx, &wl, &cq); updated || err != nil {
 			return ctrl.Result{}, err
@@ -336,8 +350,11 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 
 	// If the workload is admitted, updating the status here would set the Admitted condition to
 	// false before the workloads eviction.
+	wlOrig := wl.DeepCopy()
 	if !workload.IsAdmitted(&wl) && workload.SyncAdmittedCondition(&wl, r.clock.Now()) {
-		if err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock); err != nil {
+		if err := workload.PatchAdmissionStatus(ctx, r.client, wlOrig, true, r.clock, func() (*kueue.Workload, bool, error) {
+			return &wl, true, nil
+		}); err != nil {
 			return ctrl.Result{}, client.IgnoreNotFound(err)
 		}
 		if workload.IsAdmitted(&wl) {
@@ -388,26 +405,30 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	switch {
 	case !lqExists:
 		log.V(3).Info("Workload is inadmissible because of missing LocalQueue", "localQueue", klog.KRef(wl.Namespace, string(wl.Spec.QueueName)))
-		if workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s doesn't exist", wl.Spec.QueueName), r.clock.Now()) {
-			err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock)
+		if err := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			return &wl, workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s doesn't exist", wl.Spec.QueueName), r.clock.Now()), nil
+		}); err != nil {
 			return ctrl.Result{}, client.IgnoreNotFound(err)
 		}
 	case !lqActive:
 		log.V(3).Info("Workload is inadmissible because of stopped LocalQueue", "localQueue", klog.KRef(wl.Namespace, string(wl.Spec.QueueName)))
-		if workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s is inactive", wl.Spec.QueueName), r.clock.Now()) {
-			err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock)
+		if err := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			return &wl, workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s is inactive", wl.Spec.QueueName), r.clock.Now()), nil
+		}); err != nil {
 			return ctrl.Result{}, client.IgnoreNotFound(err)
 		}
 	case !cqOk:
 		log.V(3).Info("Workload is inadmissible because of missing ClusterQueue", "clusterQueue", klog.KRef("", string(cqName)))
-		if workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s doesn't exist", cqName), r.clock.Now()) {
-			err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock)
+		if err := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			return &wl, workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s doesn't exist", cqName), r.clock.Now()), nil
+		}); err != nil {
 			return ctrl.Result{}, client.IgnoreNotFound(err)
 		}
 	case !r.cache.ClusterQueueActive(cqName):
 		log.V(3).Info("Workload is inadmissible because ClusterQueue is inactive", "clusterQueue", klog.KRef("", string(cqName)))
-		if workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s is inactive", cqName), r.clock.Now()) {
-			err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock)
+		if err := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			return &wl, workload.UnsetQuotaReservationWithCondition(&wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s is inactive", cqName), r.clock.Now()), nil
+		}); err != nil {
 			return ctrl.Result{}, client.IgnoreNotFound(err)
 		}
 	}
@@ -444,9 +465,12 @@ func (r *WorkloadReconciler) reconcileMaxExecutionTime(ctx context.Context, wl *
 	}
 
 	if !apimeta.IsStatusConditionTrue(wl.Status.Conditions, kueue.WorkloadDeactivationTarget) {
-		workload.SetDeactivationTarget(wl, kueue.WorkloadMaximumExecutionTimeExceeded, "exceeding the maximum execution time")
-		wl.Status.AccumulatedPastExexcutionTimeSeconds = nil
-		if err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock); err != nil {
+		err := workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			workload.SetDeactivationTarget(wl, kueue.WorkloadMaximumExecutionTimeExceeded, "exceeding the maximum execution time")
+			wl.Status.AccumulatedPastExexcutionTimeSeconds = nil
+			return wl, true, nil
+		})
+		if err != nil {
 			return 0, err
 		}
 		r.recorder.Eventf(wl, corev1.EventTypeWarning, kueue.WorkloadMaximumExecutionTimeExceeded, "The maximum execution time (%ds) exceeded", *wl.Spec.MaximumExecutionTimeSeconds)
@@ -467,8 +491,11 @@ func (r *WorkloadReconciler) reconcileCheckBasedEviction(ctx context.Context, wl
 		for _, check := range workload.RejectedChecks(wl) {
 			rejectedCheckNames = append(rejectedCheckNames, check.Name)
 		}
-		workload.SetDeactivationTarget(wl, kueue.WorkloadEvictedByAdmissionCheck, fmt.Sprintf("Admission check(s): %v, were rejected", stringsutils.Join(rejectedCheckNames, ",")))
-		if err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock); err != nil {
+		err := workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			workload.SetDeactivationTarget(wl, kueue.WorkloadEvictedByAdmissionCheck, fmt.Sprintf("Admission check(s): %v, were rejected", stringsutils.Join(rejectedCheckNames, ",")))
+			return wl, true, nil
+		})
+		if err != nil {
 			return false, err
 		}
 		log.V(3).Info("Workload is evicted due to rejected admission checks", "workload", klog.KObj(wl), "rejectedChecks", rejectedCheckNames)
@@ -517,14 +544,18 @@ func (r *WorkloadReconciler) reconcileOnLocalQueueActiveState(ctx context.Contex
 
 	if !lqExists || !lq.DeletionTimestamp.IsZero() {
 		log.V(3).Info("Workload is inadmissible because the LocalQueue is terminating or missing", "localQueue", klog.KRef("", string(wl.Spec.QueueName)))
-		_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s is terminating or missing", wl.Spec.QueueName), r.clock.Now())
-		return true, workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
+		return true, workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s is terminating or missing", wl.Spec.QueueName), r.clock.Now())
+			return wl, true, nil
+		})
 	}
 
 	if queueStopPolicy != kueue.None {
 		log.V(3).Info("Workload is inadmissible because the LocalQueue is stopped", "localQueue", klog.KRef("", string(wl.Spec.QueueName)))
-		_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s is stopped", wl.Spec.QueueName), r.clock.Now())
-		return true, workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
+		return true, workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("LocalQueue %s is stopped", wl.Spec.QueueName), r.clock.Now())
+			return wl, true, nil
+		})
 	}
 
 	return false, nil
@@ -557,14 +588,18 @@ func (r *WorkloadReconciler) reconcileOnClusterQueueActiveState(ctx context.Cont
 
 	if !cqExists || !cq.DeletionTimestamp.IsZero() {
 		log.V(3).Info("Workload is inadmissible because the ClusterQueue is terminating or missing", "clusterQueue", klog.KRef("", string(cqName)))
-		_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s is terminating or missing", cqName), r.clock.Now())
-		return true, workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
+		return true, workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s is terminating or missing", cqName), r.clock.Now())
+			return wl, true, nil
+		})
 	}
 
 	if queueStopPolicy != kueue.None {
 		log.V(3).Info("Workload is inadmissible because the ClusterQueue is stopped", "clusterQueue", klog.KRef("", string(cqName)))
-		_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s is stopped", cqName), r.clock.Now())
-		return true, workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
+		return true, workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			_ = workload.UnsetQuotaReservationWithCondition(wl, kueue.WorkloadInadmissible, fmt.Sprintf("ClusterQueue %s is stopped", cqName), r.clock.Now())
+			return wl, true, nil
+		})
 	}
 
 	return false, nil
@@ -666,9 +701,11 @@ func (r *WorkloadReconciler) triggerDeactivationOrBackoffRequeue(ctx context.Con
 	}
 	// If requeuingBackoffLimitCount equals to null, the workloads is repeatedly and endless re-queued.
 	if r.waitForPodsReady.requeuingBackoffLimitCount != nil && ptr.Deref(wl.Status.RequeueState.Count, 0)+1 > *r.waitForPodsReady.requeuingBackoffLimitCount {
-		workload.SetDeactivationTarget(wl, kueue.WorkloadRequeuingLimitExceeded,
-			"exceeding the maximum number of re-queuing retries")
-		if err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock); err != nil {
+		if err := workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			workload.SetDeactivationTarget(wl, kueue.WorkloadRequeuingLimitExceeded,
+				"exceeding the maximum number of re-queuing retries")
+			return wl, true, nil
+		}); err != nil {
 			return false, err
 		}
 		return true, nil
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index 960241afd..8187d7568 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -533,11 +533,13 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 		if workload.HasQuotaReservation(wl) {
 			if !job.IsActive() {
 				log.V(6).Info("The job is no longer active, clear the workloads admission")
-				// The requeued condition status set to true only on EvictedByPreemption
-				setRequeued := evCond.Reason == kueue.WorkloadEvictedByPreemption
-				workload.SetRequeuedCondition(wl, evCond.Reason, evCond.Message, setRequeued)
-				_ = workload.UnsetQuotaReservationWithCondition(wl, "Pending", evCond.Message, r.clock.Now())
-				err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
+				err := workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+					// The requeued condition status set to true only on EvictedByPreemption
+					setRequeued := evCond.Reason == kueue.WorkloadEvictedByPreemption
+					workload.SetRequeuedCondition(wl, evCond.Reason, evCond.Message, setRequeued)
+					_ = workload.UnsetQuotaReservationWithCondition(wl, "Pending", evCond.Message, r.clock.Now())
+					return wl, true, nil
+				})
 				if err != nil {
 					return ctrl.Result{}, fmt.Errorf("clearing admission: %w", err)
 				}
diff --git a/pkg/controller/jobs/kubeflow/trainjob/trainjob_controller.go b/pkg/controller/jobs/kubeflow/trainjob/trainjob_controller.go
index d705577a6..b6136dbe4 100644
--- a/pkg/controller/jobs/kubeflow/trainjob/trainjob_controller.go
+++ b/pkg/controller/jobs/kubeflow/trainjob/trainjob_controller.go
@@ -188,6 +188,9 @@ func (t *TrainJob) RunWithPodSetsInfo(podSetsInfo []podset.PodSetInfo) error {
 				{Name: string(info.Name)},
 			},
 			// TODO: Set the labels/annotations when supported. See https://github.com/kubeflow/trainer/pull/2785
+			//
+			// NOTE: Due to the issue above, in TAS mode, missing PodSet-specific labels and annotations
+			//       prevent removal of the scheduling gate, leaving the Pod in a Pending state.
 			NodeSelector:    info.NodeSelector,
 			Tolerations:     info.Tolerations,
 			SchedulingGates: info.SchedulingGates,
diff --git a/pkg/controller/tas/node_failure_controller.go b/pkg/controller/tas/node_failure_controller.go
index 484629986..cbcf034c1 100644
--- a/pkg/controller/tas/node_failure_controller.go
+++ b/pkg/controller/tas/node_failure_controller.go
@@ -335,18 +335,22 @@ func (r *nodeFailureReconciler) handleHealthyNode(ctx context.Context, nodeName
 
 func (r *nodeFailureReconciler) removeUnhealthyNodes(ctx context.Context, wl *kueue.Workload, nodeName string) error {
 	if workload.HasUnhealthyNode(wl, nodeName) {
-		wl.Status.UnhealthyNodes = slices.DeleteFunc(wl.Status.UnhealthyNodes, func(n kueue.UnhealthyNode) bool {
-			return n.Name == nodeName
+		return workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			wl.Status.UnhealthyNodes = slices.DeleteFunc(wl.Status.UnhealthyNodes, func(n kueue.UnhealthyNode) bool {
+				return n.Name == nodeName
+			})
+			return wl, true, nil
 		})
-		return workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
 	}
 	return nil
 }
 
 func (r *nodeFailureReconciler) addUnhealthyNode(ctx context.Context, wl *kueue.Workload, nodeName string) error {
 	if !workload.HasUnhealthyNode(wl, nodeName) {
-		wl.Status.UnhealthyNodes = append(wl.Status.UnhealthyNodes, kueue.UnhealthyNode{Name: nodeName})
-		return workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
+		return workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			wl.Status.UnhealthyNodes = append(wl.Status.UnhealthyNodes, kueue.UnhealthyNode{Name: nodeName})
+			return wl, true, nil
+		})
 	}
 	return nil
 }
diff --git a/pkg/controller/workloaddispatcher/incrementaldispatcher.go b/pkg/controller/workloaddispatcher/incrementaldispatcher.go
index 1618930bc..941359987 100644
--- a/pkg/controller/workloaddispatcher/incrementaldispatcher.go
+++ b/pkg/controller/workloaddispatcher/incrementaldispatcher.go
@@ -148,8 +148,10 @@ func (r *IncrementalDispatcherReconciler) nominateWorkers(ctx context.Context, w
 	}
 
 	nominatedWorkers := append(wl.Status.NominatedClusterNames, nextNominatedWorkers...)
-	wl.Status.NominatedClusterNames = nominatedWorkers
-	if err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock); err != nil {
+	if err = workload.PatchAdmissionStatus(ctx, r.client, wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+		wl.Status.NominatedClusterNames = nominatedWorkers
+		return wl, true, nil
+	}); err != nil {
 		log.V(2).Error(err, "Failed to patch nominated clusters")
 		return reconcile.Result{}, err
 	}
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 1e3745430..f56725cd4 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -84,7 +84,7 @@ type Scheduler struct {
 	schedulingCycle int64
 
 	// Stubs.
-	applyAdmission func(context.Context, *kueue.Workload) error
+	patchAdmission func(ctx context.Context, wl *kueue.Workload) error
 }
 
 type options struct {
@@ -150,7 +150,7 @@ func New(queues *qcache.Manager, cache *schdcache.Cache, cl client.Client, recor
 		clock:                   options.clock,
 		admissionFairSharing:    options.admissionFairSharing,
 	}
-	s.applyAdmission = s.applyAdmissionWithSSA
+	s.patchAdmission = s.patchAdmissionStatus
 	return s
 }
 
@@ -314,8 +314,10 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 			// If WaitForPodsReady is enabled and WaitForPodsReady.BlockAdmission is true
 			// Block admission until all currently admitted workloads are in
 			// PodsReady condition if the waitForPodsReady is enabled
-			workload.UnsetQuotaReservationWithCondition(e.Obj, "Waiting", "waiting for all admitted workloads to be in PodsReady condition", s.clock.Now())
-			if err := workload.ApplyAdmissionStatus(ctx, s.client, e.Obj, false, s.clock); err != nil {
+			if err := workload.PatchAdmissionStatus(ctx, s.client, e.Obj, false, s.clock, func() (*kueue.Workload, bool, error) {
+				workload.UnsetQuotaReservationWithCondition(e.Obj, "Waiting", "waiting for all admitted workloads to be in PodsReady condition", s.clock.Now())
+				return e.Obj, true, nil
+			}); err != nil {
 				log.Error(err, "Could not update Workload status")
 			}
 			s.cache.WaitForPodsReady(ctx)
@@ -630,7 +632,7 @@ func (s *Scheduler) admit(ctx context.Context, e *entry, cq *schdcache.ClusterQu
 	}
 
 	s.admissionRoutineWrapper.Run(func() {
-		err := s.applyAdmission(ctx, newWorkload)
+		err := s.patchAdmission(ctx, newWorkload)
 		if err == nil {
 			// Record metrics and events for quota reservation and admission
 			s.recordWorkloadAdmissionMetrics(newWorkload, e.Obj, admission)
@@ -656,8 +658,10 @@ func (s *Scheduler) admit(ctx context.Context, e *entry, cq *schdcache.ClusterQu
 	return nil
 }
 
-func (s *Scheduler) applyAdmissionWithSSA(ctx context.Context, w *kueue.Workload) error {
-	return workload.ApplyAdmissionStatus(ctx, s.client, w, false, s.clock)
+func (s *Scheduler) patchAdmissionStatus(ctx context.Context, w *kueue.Workload) error {
+	return workload.PatchAdmissionStatus(ctx, s.client, w, false, s.clock, func() (*kueue.Workload, bool, error) {
+		return w, true, nil
+	})
 }
 
 type entryOrdering struct {
@@ -766,15 +770,14 @@ func (s *Scheduler) requeueAndUpdate(ctx context.Context, e entry) {
 
 	added := s.queues.RequeueWorkload(ctx, &e.Info, e.requeueReason)
 	log.V(2).Info("Workload re-queued", "workload", klog.KObj(e.Obj), "clusterQueue", klog.KRef("", string(e.ClusterQueue)), "queue", klog.KRef(e.Obj.Namespace, string(e.Obj.Spec.QueueName)), "requeueReason", e.requeueReason, "added", added, "status", e.status)
-
 	if e.status == notNominated || e.status == skipped {
-		patch := workload.PrepareWorkloadPatch(e.Obj, true, s.clock)
-		reservationIsChanged := workload.UnsetQuotaReservationWithCondition(patch, "Pending", e.inadmissibleMsg, s.clock.Now())
-		resourceRequestsIsChanged := workload.PropagateResourceRequests(patch, &e.Info)
-		if reservationIsChanged || resourceRequestsIsChanged {
-			if err := workload.ApplyAdmissionStatusPatch(ctx, s.client, patch); err != nil {
-				log.Error(err, "Could not update Workload status")
-			}
+		if err := workload.PatchAdmissionStatus(ctx, s.client, e.Obj, true, s.clock, func() (*kueue.Workload, bool, error) {
+			wl := e.Obj.DeepCopy()
+			reservationIsChanged := workload.UnsetQuotaReservationWithCondition(wl, "Pending", e.inadmissibleMsg, s.clock.Now())
+			resourceRequestsIsChanged := workload.PropagateResourceRequests(wl, &e.Info)
+			return wl, reservationIsChanged || resourceRequestsIsChanged, nil
+		}); 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/workload/workload.go b/pkg/workload/workload.go
index f3261896e..714365f07 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -896,6 +896,17 @@ func ApplyAdmissionStatusPatch(ctx context.Context, c client.Client, patch *kueu
 	return c.Status().Patch(ctx, patch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)
 }
 
+// PatchAdmissionStatus updates the admission status of a workload.
+// It runs the update function and, if updated, applies the SSA Patch status.
+func PatchAdmissionStatus(ctx context.Context, c client.Client, w *kueue.Workload, strict bool, clk clock.Clock, update func() (*kueue.Workload, bool, error)) error {
+	wPatched, updated, err := update()
+	if err != nil || !updated {
+		return err
+	}
+
+	return ApplyAdmissionStatus(ctx, c, wPatched, strict, clk)
+}
+
 type Ordering struct {
 	PodsReadyRequeuingTimestamp config.RequeuingTimestamp
 }
@@ -1110,7 +1121,9 @@ func Evict(ctx context.Context, c client.Client, recorder record.EventRecorder,
 	}
 	prepareForEviction(wl, clock.Now(), evictionReason, msg)
 	reportWorkloadEvictedOnce := workloadEvictionStateInc(wl, reason, underlyingCause)
-	if err := ApplyAdmissionStatus(ctx, c, wl, true, clock); err != nil {
+	if err := PatchAdmissionStatus(ctx, c, wl, true, clock, func() (*kueue.Workload, bool, error) {
+		return wl, true, nil
+	}); err != nil {
 		return err
 	}
 	if wl.Status.Admission == nil {
diff --git a/site/content/zh-CN/docs/tasks/run/deployment.md b/site/content/zh-CN/docs/tasks/run/deployment.md
index 2552ca906..eae9b3ad0 100644
--- a/site/content/zh-CN/docs/tasks/run/deployment.md
+++ b/site/content/zh-CN/docs/tasks/run/deployment.md
@@ -1,40 +1,38 @@
 ---
-title: "Run Deployment"
+title: "运行 Deployment"
 linkTitle: "Deployment"
 date: 2024-07-25
 weight: 6
 description: >
-  Run a Deployment as a Kueue-managed workload.
+  将 Deployment 作为 Kueue 管理的工作负载运行。
 ---
 
-This page shows how to leverage Kueue's scheduling and resource management
-capabilities when running Deployments.
-Although Kueue does not yet support managing a Deployment as a single Workload, 
-it's still possible to leverage Kueue's scheduling and resource management capabilities for the individual Pods of the Deployment.
+本页面展示了在运行 Deployment 时如何利用 Kueue 的调度和资源管理能力。
+虽然 Kueue 尚不支持将 Deployment 作为单个工作负载进行管理，
+但仍然可以利用 Kueue 的调度和资源管理能力来管理 Deployment 中的各个 Pod。
 
-We demonstrate how to support scheduling Deployments in Kueue based on the Plain Pod integration,
-where every Pod from a Deployment is represented as a single independent Plain Pod.
-This approach allows independent resource management for the Pods, and thus scale-out and scale-in of the Deployment.
+我们演示了如何基于 Plain Pod 集成在 Kueue 中支持调度 Deployment，
+其中 Deployment 中的每个 Pod 都被表示为一个独立的 Plain Pod。
+这种方法允许对 Pod 进行独立的资源管理，从而实现 Deployment 的扩容和缩容。
 
-This guide is for [serving users](/docs/tasks#serving-user) that have a basic understanding of Kueue.
-For more information, see [Kueue's overview](/docs/overview).
+本指南适用于对 Kueue 有基本了解的[服务用户](/zh-CN/docs/tasks#serving-user)。
+更多信息，请参见 [Kueue 概述](/zh-CN/docs/overview)。
 
-## Before you begin
+## 开始之前
 
-1. Learn how to [install Kueue with a custom manager configuration](/docs/installation/#install-a-custom-configured-released-version).
+1. 学习如何[使用自定义管理器配置来安装 Kueue](/docs/installation/#install-a-custom-configured-released-version)。
 
-2. Follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin)
-to learn how to enable and configure the `pod` integration.
+2. 按照[运行 Plain Pod](/zh-CN/docs/tasks/run/plain_pods/#before-you-begin)中的步骤学习如何启用和配置 `pod` 集成。
 
-3. Check [Administer cluster quotas](/docs/tasks/manage/administer_cluster_quotas) for details on the initial Kueue setup.
+3. 查看[管理集群配额](/zh-CN/docs/tasks/manage/administer_cluster_quotas)了解初始 Kueue 设置的详细信息。
 
-## Running a Deployment admitted by Kueue
+## 运行由 Kueue 准入的 Deployment
 
-When running Deployment on Kueue, take into consideration the following aspects:
+在 Kueue 上运行 Deployment 时，需要考虑以下方面：
 
-### a. Queue selection
+### a. 队列选择
 
-The target [local queue](/docs/concepts/local_queue) should be specified in the `metadata.labels` section of the Deployment configuration.
+目标[本地队列](/zh-CN/docs/concepts/local_queue)应在 Deployment 配置的 `metadata.labels` 部分中指定。
 
 ```yaml
 metadata:
@@ -42,9 +40,9 @@ metadata:
     kueue.x-k8s.io/queue-name: user-queue
 ```
 
-### b. Configure the resource needs
+### b. 配置资源需求
 
-The resource needs of the workload can be configured in the `spec.template.spec.containers`.
+工作负载的资源需求可以在 `spec.template.spec.containers` 中配置。
 
 ```yaml
     - resources:
@@ -52,28 +50,28 @@ The resource needs of the workload can be configured in the `spec.template.spec.
           cpu: 3
 ```
 
-### c. Scaling
+### c. 扩缩容
 
-You may perform scale up or scale down operations on Deployments.
-On scale-in, the excess Pods are deleted, and the quota is freed.
-On scale-out, new Pods are created, and remain suspended until their corresponding workloads get admitted.
-If there is not enough quota in your cluster, the Deployment might run only a subset of Pods. 
-So, if your workloads are business-critical, 
-you can consider reserving the quota only for the serving workloads by the ClusterQueue `lendingLimit`. 
-The `lendingLimit` allows you to rapidly scale out the critical serving workload.
-For more `lendingLimit` details, please see the [ClusterQueue page](docs/concepts/cluster_queue#lendinglimit).
+您可以对 Deployment 执行扩容或缩容操作。
+在缩容时，多余的 Pod 会被删除，配额会被释放。
+在扩容时，会创建新的 Pod，这些 Pod 会保持挂起状态，直到相应的工作负载被准入。
+如果集群中没有足够的配额，Deployment 可能只运行部分 Pod。
+因此，如果您的工作负载对业务至关重要，
+您可以考虑通过 ClusterQueue 的 `lendingLimit` 仅为服务工作负载保留配额。
+`lendingLimit` 允许您快速扩容关键的服务工作负载。
+有关 `lendingLimit` 的更多详细信息，请参见 [ClusterQueue 页面](/zh-CN/docs/concepts/cluster_queue#lendinglimit)。
 
-### d. Limitations
+### d. 限制
 
-- The scope for Deployments is implied by the pod integration's namespace selector. There's no independent control for deployments.
+- Deployment 的范围由 `pod` 集成的命名空间选择器所默示。不能独立控制 Deployment。
 
-## Example
+## 示例
 
-Here is a sample Deployment:
+以下是一个示例 Deployment：
 
 {{< include "examples/serving-workloads/sample-deployment.yaml" "yaml" >}}
 
-You can create the Deployment using the following command:
+您可以使用以下命令创建 Deployment：
 ```sh
 kubectl create -f sample-deployment.yaml
 ```



Test patch:
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 6ead85840..f69ea2cdb 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -3866,7 +3866,7 @@ func TestSchedule(t *testing.T) {
 
 			gotScheduled := make(map[workload.Reference]kueue.Admission)
 			var mu sync.Mutex
-			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+			scheduler.patchAdmission = func(ctx context.Context, w *kueue.Workload) error {
 				if tc.admissionError != nil {
 					return tc.admissionError
 				}
@@ -4752,7 +4752,7 @@ func TestLastSchedulingContext(t *testing.T) {
 			scheduler := New(qManager, cqCache, cl, recorder, WithClock(t, fakeClock))
 			gotScheduled := make(map[workload.Reference]kueue.Admission)
 			var mu sync.Mutex
-			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+			scheduler.patchAdmission = func(ctx context.Context, w *kueue.Workload) error {
 				mu.Lock()
 				gotScheduled[workload.Key(w)] = *w.Status.Admission
 				mu.Unlock()
@@ -7577,7 +7577,7 @@ func TestScheduleForTAS(t *testing.T) {
 			scheduler := New(qManager, cqCache, cl, recorder)
 			gotScheduled := make([]workload.Reference, 0)
 			var mu sync.Mutex
-			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+			scheduler.patchAdmission = func(ctx context.Context, w *kueue.Workload) error {
 				mu.Lock()
 				gotScheduled = append(gotScheduled, workload.Key(w))
 				mu.Unlock()
@@ -8113,7 +8113,7 @@ func TestScheduleForTASPreemption(t *testing.T) {
 			scheduler := New(qManager, cqCache, cl, recorder)
 			gotScheduled := make([]workload.Reference, 0)
 			var mu sync.Mutex
-			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+			scheduler.patchAdmission = func(ctx context.Context, w *kueue.Workload) error {
 				mu.Lock()
 				gotScheduled = append(gotScheduled, workload.Key(w))
 				mu.Unlock()
@@ -9085,7 +9085,7 @@ func TestScheduleForTASCohorts(t *testing.T) {
 			scheduler := New(qManager, cqCache, cl, recorder)
 			gotScheduled := make([]workload.Reference, 0)
 			var mu sync.Mutex
-			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+			scheduler.patchAdmission = func(ctx context.Context, w *kueue.Workload) error {
 				mu.Lock()
 				gotScheduled = append(gotScheduled, workload.Key(w))
 				mu.Unlock()
@@ -9435,7 +9435,7 @@ func TestScheduleForAFS(t *testing.T) {
 
 			gotScheduled := make(map[workload.Reference]kueue.Admission)
 			var mu sync.Mutex
-			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+			scheduler.patchAdmission = func(ctx context.Context, w *kueue.Workload) error {
 				mu.Lock()
 				gotScheduled[workload.Key(w)] = *w.Status.Admission
 				mu.Unlock()
diff --git a/pkg/workload/workload_test.go b/pkg/workload/workload_test.go
index e3bbeea73..e92590b5b 100644
--- a/pkg/workload/workload_test.go
+++ b/pkg/workload/workload_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package workload
 
 import (
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -1509,3 +1510,59 @@ func TestSetQuotaReservation(t *testing.T) {
 		})
 	}
 }
+func TestPatchAdmissionStatus(t *testing.T) {
+	now := time.Now()
+	fakeClock := testingclock.NewFakeClock(now)
+	type patchCall struct {
+		updated bool
+		err     error
+	}
+
+	errUpdate := errors.New("update error")
+
+	tests := map[string]struct {
+		patchCall patchCall
+		wantErr   error
+	}{
+		"update returns true": {
+			patchCall: patchCall{updated: true, err: nil},
+			wantErr:   nil,
+		},
+		"update returns false": {
+			patchCall: patchCall{updated: false, err: nil},
+			wantErr:   nil,
+		},
+		"update returns error": {
+			patchCall: patchCall{updated: false, err: errUpdate},
+			wantErr:   errUpdate,
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			ctx, _ := utiltesting.ContextWithLog(t)
+			wl := utiltesting.MakeWorkload("foo", "default").Obj()
+			cl := utiltesting.NewFakeClientSSAAsSM(wl)
+			called := false
+			gotErr := PatchAdmissionStatus(
+				ctx,
+				cl,
+				wl,
+				true,
+				fakeClock,
+				func() (*kueue.Workload, bool, error) {
+					called = true
+					return wl, tc.patchCall.updated, tc.patchCall.err
+				},
+			)
+			if diff := cmp.Diff(tc.wantErr, gotErr, cmpopts.EquateErrors()); diff != "" {
+				t.Errorf("Unexpected error (-want/+got)\n%s", diff)
+			}
+			if !called {
+				t.Errorf("expected update func to be called when feature enabled")
+			}
+			if tc.patchCall.updated && !called {
+				t.Errorf("expected update func to be called when feature disabled and update true")
+			}
+		})
+	}
+}
diff --git a/test/integration/multikueue/dispatcher_test.go b/test/integration/multikueue/dispatcher_test.go
index 1e79bbbd8..fbf6eff69 100644
--- a/test/integration/multikueue/dispatcher_test.go
+++ b/test/integration/multikueue/dispatcher_test.go
@@ -32,7 +32,6 @@ import (
 
 	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
-	"sigs.k8s.io/kueue/pkg/constants"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
@@ -217,13 +216,14 @@ var _ = ginkgo.Describe("MultiKueueDispatcherIncremental", ginkgo.Ordered, ginkg
 		ginkgo.By("setting the check conditions for eviction", func() {
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
-				wlPatch := workload.PrepareWorkloadPatch(createdWorkload, true, realClock)
-				workload.SetAdmissionCheckState(&wlPatch.Status.AdmissionChecks, kueue.AdmissionCheckState{
-					Name:    kueue.AdmissionCheckReference(multiKueueAC.Name),
-					State:   kueue.CheckStateRejected,
-					Message: "check rejected",
-				}, realClock)
-				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+				g.Expect(workload.PatchAdmissionStatus(managerTestCluster.ctx, managerTestCluster.client, createdWorkload, true, realClock, func() (*kueue.Workload, bool, error) {
+					workload.SetAdmissionCheckState(&createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckState{
+						Name:    kueue.AdmissionCheckReference(multiKueueAC.Name),
+						State:   kueue.CheckStateRejected,
+						Message: "check rejected",
+					}, realClock)
+					return createdWorkload, true, nil
+				})).To(gomega.Succeed())
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
@@ -391,9 +391,10 @@ var _ = ginkgo.Describe("MultiKueueDispatcherExternal", ginkgo.Ordered, ginkgo.C
 			gomega.Eventually(func(g gomega.Gomega) {
 				managerWl := &kueue.Workload{}
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
-				wlPatch := workload.PrepareWorkloadPatch(managerWl, true, realClock)
-				wlPatch.Status.NominatedClusterNames = []string{workerCluster2.Name}
-				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+				g.Expect(workload.PatchAdmissionStatus(managerTestCluster.ctx, managerTestCluster.client, managerWl, true, realClock, func() (*kueue.Workload, bool, error) {
+					managerWl.Status.NominatedClusterNames = []string{workerCluster2.Name}
+					return managerWl, true, nil
+				})).To(gomega.Succeed())
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
@@ -420,9 +421,10 @@ var _ = ginkgo.Describe("MultiKueueDispatcherExternal", ginkgo.Ordered, ginkgo.C
 			managerWl := &kueue.Workload{}
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
-				wlPatch := workload.PrepareWorkloadPatch(managerWl, true, realClock)
-				wlPatch.Status.NominatedClusterNames = []string{workerCluster1.Name, workerCluster2.Name}
-				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+				g.Expect(workload.PatchAdmissionStatus(managerTestCluster.ctx, managerTestCluster.client, managerWl, true, realClock, func() (*kueue.Workload, bool, error) {
+					managerWl.Status.NominatedClusterNames = []string{workerCluster1.Name, workerCluster2.Name}
+					return managerWl, true, nil
+				})).To(gomega.Succeed())
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
@@ -461,13 +463,14 @@ var _ = ginkgo.Describe("MultiKueueDispatcherExternal", ginkgo.Ordered, ginkgo.C
 		ginkgo.By("setting the check conditions for eviction", func() {
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
-				wlPatch := workload.PrepareWorkloadPatch(createdWorkload, true, realClock)
-				workload.SetAdmissionCheckState(&wlPatch.Status.AdmissionChecks, kueue.AdmissionCheckState{
-					Name:    kueue.AdmissionCheckReference(multiKueueAC.Name),
-					State:   kueue.CheckStateRejected,
-					Message: "check rejected",
-				}, realClock)
-				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+				g.Expect(workload.PatchAdmissionStatus(managerTestCluster.ctx, managerTestCluster.client, createdWorkload, true, realClock, func() (*kueue.Workload, bool, error) {
+					workload.SetAdmissionCheckState(&createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckState{
+						Name:    kueue.AdmissionCheckReference(multiKueueAC.Name),
+						State:   kueue.CheckStateRejected,
+						Message: "check rejected",
+					}, realClock)
+					return createdWorkload, true, nil
+				})).To(gomega.Succeed())
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
@@ -679,9 +682,10 @@ var _ = ginkgo.Describe("MultiKueueDispatcherAllAtOnce", ginkgo.Ordered, ginkgo.
 		ginkgo.By("Resetting ClusterName field in the manager", func() {
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
-				wlPatch := workload.PrepareWorkloadPatch(createdWorkload, true, realClock)
-				wlPatch.Status.ClusterName = nil
-				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+				g.Expect(workload.PatchAdmissionStatus(managerTestCluster.ctx, managerTestCluster.client, createdWorkload, true, realClock, func() (*kueue.Workload, bool, error) {
+					createdWorkload.Status.ClusterName = nil
+					return createdWorkload, true, nil
+				})).To(gomega.Succeed())
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
diff --git a/test/integration/singlecluster/controller/core/workload_controller_test.go b/test/integration/singlecluster/controller/core/workload_controller_test.go
index a3cf38061..27949fa6a 100644
--- a/test/integration/singlecluster/controller/core/workload_controller_test.go
+++ b/test/integration/singlecluster/controller/core/workload_controller_test.go
@@ -500,8 +500,10 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
 			ginkgo.By("evicting the workload, the accumulated admission time is updated", func() {
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, key, wl)).To(gomega.Succeed())
-					workload.SetEvictedCondition(wl, "ByTest", "by test")
-					g.Expect(workload.ApplyAdmissionStatus(ctx, k8sClient, wl, false, realClock)).To(gomega.Succeed())
+					g.Expect(workload.PatchAdmissionStatus(ctx, k8sClient, wl, true, realClock, func() (*kueue.Workload, bool, error) {
+						workload.SetEvictedCondition(wl, "ByTest", "by test")
+						return wl, true, nil
+					})).Should(gomega.Succeed())
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 				util.FinishEvictionForWorkloads(ctx, k8sClient, wl)
 
diff --git a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
index e075b6e69..07df724c0 100644
--- a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
@@ -795,11 +795,14 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 
 				createdPod := &corev1.Pod{}
 				ginkgo.By("checking that the Pods get a deletion timestamp when the workload is evicted", func() {
-					gomega.Expect(func() error {
-						w := createdWorkload.DeepCopy()
-						workload.SetEvictedCondition(w, "ByTest", "by test")
-						return workload.ApplyAdmissionStatus(ctx, k8sClient, w, false, realClock)
-					}()).Should(gomega.Succeed())
+					createdWorkload := &kueue.Workload{}
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
+						g.Expect(workload.PatchAdmissionStatus(ctx, k8sClient, createdWorkload, true, realClock, func() (*kueue.Workload, bool, error) {
+							workload.SetEvictedCondition(createdWorkload, "ByTest", "by test")
+							return createdWorkload, true, nil
+						})).Should(gomega.Succeed())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 					gomega.Eventually(func(g gomega.Gomega) {
 						g.Expect(k8sClient.Get(ctx, pod1LookupKey, createdPod)).To(gomega.Succeed())
@@ -1387,10 +1390,13 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 				})
 
 				ginkgo.By("setting evicted condition to true", func() {
-					workload.SetEvictedCondition(wl, kueue.WorkloadEvictedByPreemption, "By test")
-					gomega.Expect(
-						workload.ApplyAdmissionStatus(ctx, k8sClient, wl, false, realClock),
-					).Should(gomega.Succeed())
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+						g.Expect(workload.PatchAdmissionStatus(ctx, k8sClient, wl, true, realClock, func() (*kueue.Workload, bool, error) {
+							workload.SetEvictedCondition(wl, kueue.WorkloadEvictedByPreemption, "By test")
+							return wl, true, nil
+						})).Should(gomega.Succeed())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
 				})
 
 				podKey := client.ObjectKeyFromObject(pod)
diff --git a/test/integration/singlecluster/scheduler/podsready/scheduler_test.go b/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
index ee5007946..59858ffae 100644
--- a/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
+++ b/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
@@ -330,8 +330,10 @@ var _ = ginkgo.Describe("SchedulerWithWaitForPodsReady", func() {
 				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(prodWl), prodWl)).Should(gomega.Succeed())
 				g.Expect(workload.IsActive(prodWl)).Should(gomega.BeFalse())
 				g.Expect(prodWl.Status.RequeueState).Should(gomega.BeNil())
-				workload.SetRequeuedCondition(prodWl, kueue.WorkloadDeactivated, "by test", false)
-				g.Expect(workload.ApplyAdmissionStatus(ctx, k8sClient, prodWl, true, realClock)).Should(gomega.Succeed())
+				g.Expect(workload.PatchAdmissionStatus(ctx, k8sClient, prodWl, true, realClock, func() (*kueue.Workload, bool, error) {
+					workload.SetRequeuedCondition(prodWl, kueue.WorkloadDeactivated, "by test", false)
+					return prodWl, true, nil
+				})).Should(gomega.Succeed())
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			util.FinishEvictionForWorkloads(ctx, k8sClient, prodWl)
 			// should observe a metrics of WorkloadEvictedByDeactivation
diff --git a/test/integration/singlecluster/webhook/core/workload_test.go b/test/integration/singlecluster/webhook/core/workload_test.go
index c2f2ae2e5..c26fb1a7f 100644
--- a/test/integration/singlecluster/webhook/core/workload_test.go
+++ b/test/integration/singlecluster/webhook/core/workload_test.go
@@ -315,14 +315,19 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 		ginkgo.DescribeTable("Should have valid values when setting Admission", func(w func() *kueue.Workload, a *kueue.Admission, errorType gomega.OmegaMatcher) {
 			wl := w()
 			util.MustCreate(ctx, k8sClient, wl)
-			gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), wl)).To(gomega.Succeed())
-			workload.SetQuotaReservation(wl, a, clock.RealClock{})
-			err := workload.ApplyAdmissionStatus(ctx, k8sClient, wl, false, clock.RealClock{})
-			if errorType != nil {
-				gomega.Expect(err).Should(errorType)
-			} else {
-				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
-			}
+
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), wl)).To(gomega.Succeed())
+				err := workload.PatchAdmissionStatus(ctx, k8sClient, wl, true, clock.RealClock{}, func() (*kueue.Workload, bool, error) {
+					workload.SetQuotaReservation(wl, a, clock.RealClock{})
+					return wl, true, nil
+				})
+				if errorType != nil {
+					g.Expect(err).Should(errorType)
+				} else {
+					g.Expect(err).ShouldNot(gomega.HaveOccurred())
+				}
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		},
 			ginkgo.Entry("invalid clusterQueue name",
 				func() *kueue.Workload {
diff --git a/test/performance/scheduler/runner/controller/controller.go b/test/performance/scheduler/runner/controller/controller.go
index 247d29ad3..788258160 100644
--- a/test/performance/scheduler/runner/controller/controller.go
+++ b/test/performance/scheduler/runner/controller/controller.go
@@ -120,8 +120,10 @@ func (r *reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	// this should only:
 	// 1. finish the workloads eviction
 	if apimeta.IsStatusConditionTrue(wl.Status.Conditions, kueue.WorkloadEvicted) {
-		_ = workload.UnsetQuotaReservationWithCondition(&wl, "Pending", "Evicted by the test runner", time.Now())
-		err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock)
+		err := workload.PatchAdmissionStatus(ctx, r.client, &wl, true, r.clock, func() (*kueue.Workload, bool, error) {
+			_ = workload.UnsetQuotaReservationWithCondition(&wl, "Pending", "Evicted by the test runner", time.Now())
+			return &wl, true, nil
+		})
 		if err == nil {
 			log.V(5).Info("Finish eviction")
 		}
diff --git a/test/util/util.go b/test/util/util.go
index 1aa368b25..6e603e28f 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -401,9 +401,11 @@ func SetRequeuedConditionWithPodsReadyTimeout(ctx context.Context, k8sClient cli
 	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
 		var wl kueue.Workload
 		g.Expect(k8sClient.Get(ctx, wlKey, &wl)).Should(gomega.Succeed())
-		workload.SetRequeuedCondition(&wl, kueue.WorkloadEvictedByPodsReadyTimeout,
-			fmt.Sprintf("Exceeded the PodsReady timeout %s", klog.KObj(&wl).String()), false)
-		g.Expect(workload.ApplyAdmissionStatus(ctx, k8sClient, &wl, true, clock.RealClock{})).Should(gomega.Succeed())
+		g.Expect(workload.PatchAdmissionStatus(ctx, k8sClient, &wl, true, clock.RealClock{}, func() (*kueue.Workload, bool, error) {
+			workload.SetRequeuedCondition(&wl, kueue.WorkloadEvictedByPodsReadyTimeout,
+				fmt.Sprintf("Exceeded the PodsReady timeout %s", klog.KObj(&wl).String()), false)
+			return &wl, true, nil
+		})).Should(gomega.Succeed())
 	}, Timeout, Interval).Should(gomega.Succeed())
 }
 
@@ -751,12 +753,14 @@ func SetQuotaReservation(ctx context.Context, k8sClient client.Client, wlKey cli
 	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
 		updatedWl := &kueue.Workload{}
 		g.ExpectWithOffset(1, k8sClient.Get(ctx, wlKey, updatedWl)).To(gomega.Succeed())
-		if admission == nil {
-			workload.UnsetQuotaReservationWithCondition(updatedWl, "EvictedByTest", "Evicted By Test", clk.Now())
-		} else {
-			workload.SetQuotaReservation(updatedWl, admission, clk)
-		}
-		g.ExpectWithOffset(1, workload.ApplyAdmissionStatus(ctx, k8sClient, updatedWl, false, clk)).To(gomega.Succeed())
+		g.ExpectWithOffset(1, workload.PatchAdmissionStatus(ctx, k8sClient, updatedWl, true, clk, func() (*kueue.Workload, bool, error) {
+			if admission == nil {
+				workload.UnsetQuotaReservationWithCondition(updatedWl, "EvictedByTest", "Evicted By Test", clk.Now())
+			} else {
+				workload.SetQuotaReservation(updatedWl, admission, clk)
+			}
+			return updatedWl, true, nil
+		})).To(gomega.Succeed())
 	}, Timeout, Interval).Should(gomega.Succeed())
 }
 
@@ -766,10 +770,13 @@ func SetQuotaReservation(ctx context.Context, k8sClient client.Client, wlKey cli
 func SyncAdmittedConditionForWorkloads(ctx context.Context, k8sClient client.Client, wls ...*kueue.Workload) {
 	var updatedWorkload kueue.Workload
 	for _, wl := range wls {
-		gomega.ExpectWithOffset(1, k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &updatedWorkload)).To(gomega.Succeed())
-		if workload.SyncAdmittedCondition(&updatedWorkload, time.Now()) {
-			gomega.ExpectWithOffset(1, workload.ApplyAdmissionStatus(ctx, k8sClient, &updatedWorkload, false, clock.RealClock{})).To(gomega.Succeed())
-		}
+		gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
+			g.ExpectWithOffset(1, k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &updatedWorkload)).To(gomega.Succeed())
+			g.ExpectWithOffset(1, workload.PatchAdmissionStatus(ctx, k8sClient, &updatedWorkload, true, clock.RealClock{}, func() (*kueue.Workload, bool, error) {
+				updated := workload.SyncAdmittedCondition(&updatedWorkload, time.Now())
+				return &updatedWorkload, updated, nil
+			})).To(gomega.Succeed())
+		}, Timeout, Interval).Should(gomega.Succeed())
 	}
 }
 
@@ -792,11 +799,11 @@ func FinishEvictionForWorkloads(ctx context.Context, k8sClient client.Client, wl
 			var updatedWorkload kueue.Workload
 			g.Expect(k8sClient.Get(ctx, key, &updatedWorkload)).Should(gomega.Succeed())
 			if apimeta.IsStatusConditionTrue(updatedWorkload.Status.Conditions, kueue.WorkloadQuotaReserved) {
-				workload.UnsetQuotaReservationWithCondition(&updatedWorkload, "Pending", "By test", time.Now())
-				g.Expect(workload.ApplyAdmissionStatus(ctx, k8sClient, &updatedWorkload, true, clock.RealClock{})).Should(
-					gomega.Succeed(),
-					fmt.Sprintf("Unable to unset quota reservation for %q", key),
-				)
+				g.Expect(workload.PatchAdmissionStatus(ctx, k8sClient, &updatedWorkload, true, clock.RealClock{}, func() (*kueue.Workload, bool, error) {
+					workload.UnsetQuotaReservationWithCondition(&updatedWorkload, "Pending", "By test", time.Now())
+					return &updatedWorkload, true, nil
+				}),
+				).Should(gomega.Succeed(), fmt.Sprintf("Unable to unset quota reservation for %q", key))
 			}
 		}, Timeout, Interval).Should(gomega.Succeed())
 	}
diff --git a/test/util/util_scheduling.go b/test/util/util_scheduling.go
index 47083177c..e0693dcb3 100644
--- a/test/util/util_scheduling.go
+++ b/test/util/util_scheduling.go
@@ -59,8 +59,11 @@ func FinishEvictionOfWorkloadsInCQ(ctx context.Context, k8sClient client.Client,
 			evicted := meta.IsStatusConditionTrue(wl.Status.Conditions, kueue.WorkloadEvicted)
 			quotaReserved := meta.IsStatusConditionTrue(wl.Status.Conditions, kueue.WorkloadQuotaReserved)
 			if evicted && quotaReserved {
-				workload.UnsetQuotaReservationWithCondition(&wl, "Pending", "Eviction finished by test", time.Now())
-				g.Expect(workload.ApplyAdmissionStatus(ctx, k8sClient, &wl, true, realClock)).To(gomega.Succeed())
+				g.Expect(workload.PatchAdmissionStatus(ctx, k8sClient, &wl, true, realClock, func() (*kueue.Workload, bool, error) {
+					workload.UnsetQuotaReservationWithCondition(&wl, "Pending", "Eviction finished by test", time.Now())
+					return &wl, true, nil
+				}),
+				).To(gomega.Succeed())
 				finished.Insert(wl.UID)
 			}
 		}