Repository: kubernetes-sigs/kueue

Issue:
MultiKueue: Update documentation related to the MultiKueue Dispatcher
I would like to have documentation on the two releated features: 
- MultiKueue dispatcher API: https://github.com/kubernetes-sigs/kueue/issues/5141
- MultiKueue incremental mode: https://github.com/kubernetes-sigs/kueue/issues/3757

The implementation of both is addressed in 0.13 under https://github.com/kubernetes-sigs/kueue/pull/5782


Hints from issue comments:
@mimowo: /kind documentation
cc @mszadkow @tenzen-y @mwielgus @mszadkow: /assign

Solution patch:
diff --git a/config/components/certmanager/certificate-metrics.yaml b/config/components/certmanager/certificate-metrics.yaml
index 2fab4da27..afdab84cc 100644
--- a/config/components/certmanager/certificate-metrics.yaml
+++ b/config/components/certmanager/certificate-metrics.yaml
@@ -6,7 +6,7 @@ metadata:
   name: metrics-certs  # this name should match the one appeared in kustomizeconfig.yaml
   namespace: system
 spec:
-  commonName: kueue-metrics
+  commonName: metrics
   dnsNames:
   # METRICS_SERVICE_NAME and METRICS_SERVICE_NAMESPACE will be substituted by kustomize
   # replacements in the config/default/kustomization.yaml file.
diff --git a/config/components/certmanager/common_name_transformer.yaml b/config/components/certmanager/common_name_transformer.yaml
new file mode 100644
index 000000000..549850126
--- /dev/null
+++ b/config/components/certmanager/common_name_transformer.yaml
@@ -0,0 +1,9 @@
+apiVersion: builtin
+kind: PrefixSuffixTransformer
+metadata:
+  name: add-prefix-to-common-name
+prefix: kueue-
+fieldSpecs:
+  - kind: Certificate
+    group: cert-manager.io
+    path: spec/commonName
diff --git a/config/components/certmanager/kustomization.yaml b/config/components/certmanager/kustomization.yaml
index fc1018507..7a4276d21 100644
--- a/config/components/certmanager/kustomization.yaml
+++ b/config/components/certmanager/kustomization.yaml
@@ -2,6 +2,9 @@ resources:
 - certificate.yaml
 - certificate-metrics.yaml
 
-
 configurations:
 - kustomizeconfig.yaml
+
+transformers:
+- common_name_transformer.yaml
+- secret_name_transformer.yaml
diff --git a/config/components/certmanager/secret_name_transformer.yaml b/config/components/certmanager/secret_name_transformer.yaml
new file mode 100644
index 000000000..b320ed302
--- /dev/null
+++ b/config/components/certmanager/secret_name_transformer.yaml
@@ -0,0 +1,9 @@
+apiVersion: builtin
+kind: PrefixSuffixTransformer
+metadata:
+  name: add-prefix-to-secret-name
+prefix: kueue-
+fieldSpecs:
+  - kind: Certificate
+    group: cert-manager.io
+    path: spec/secretName
diff --git a/config/default/cert_metrics_manager_patch.yaml b/config/default/cert_metrics_manager_patch.yaml
index 78fc5929f..d0082d114 100644
--- a/config/default/cert_metrics_manager_patch.yaml
+++ b/config/default/cert_metrics_manager_patch.yaml
@@ -14,7 +14,7 @@
   value:
     name: metrics-certs
     secret:
-      secretName: metrics-server-cert
+      secretName: kueue-metrics-server-cert
       optional: false
       items:
         - key: ca.crt
diff --git a/config/default/manager_webhook_patch.yaml b/config/default/manager_webhook_patch.yaml
index 738de350b..e8880dde0 100644
--- a/config/default/manager_webhook_patch.yaml
+++ b/config/default/manager_webhook_patch.yaml
@@ -20,4 +20,4 @@ spec:
       - name: cert
         secret:
           defaultMode: 420
-          secretName: webhook-server-cert
+          secretName: kueue-webhook-server-cert
diff --git a/pkg/cache/snapshot.go b/pkg/cache/snapshot.go
index 433a400d2..4cd01c3a4 100644
--- a/pkg/cache/snapshot.go
+++ b/pkg/cache/snapshot.go
@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -182,6 +183,7 @@ func (c *Cache) Snapshot(ctx context.Context, options ...SnapshotOption) (*Snaps
 // snapshotClusterQueue creates a copy of ClusterQueue that includes
 // references to immutable objects and deep copies of changing ones.
 func (c *Cache) snapshotClusterQueue(ctx context.Context, cq *clusterQueue, afsEntryPenalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) (*ClusterQueueSnapshot, error) {
+	log := log.FromContext(ctx)
 	cc := &ClusterQueueSnapshot{
 		Name:                          cq.Name,
 		ResourceGroups:                make([]ResourceGroup, len(cq.ResourceGroups)),
@@ -215,6 +217,7 @@ func (c *Cache) snapshotClusterQueue(ctx context.Context, cq *clusterQueue, afsE
 				return nil, fmt.Errorf("failed to calculate LocalQueue FS usage for LocalQueue %v", client.ObjectKey{Namespace: wl.Obj.Namespace, Name: string(wl.Obj.Spec.QueueName)})
 			}
 			wl.LocalQueueFSUsage = &usage
+			log.V(3).Info("Calculated LocalQueueFSUsage for workload", "workload", klog.KObj(wl.Obj), "queue", wl.Obj.Spec.QueueName, "usage", usage)
 		}
 	}
 	return cc, nil
diff --git a/pkg/controller/admissionchecks/provisioning/controller.go b/pkg/controller/admissionchecks/provisioning/controller.go
index bfabb5199..7feca94b8 100644
--- a/pkg/controller/admissionchecks/provisioning/controller.go
+++ b/pkg/controller/admissionchecks/provisioning/controller.go
@@ -499,6 +499,7 @@ func (c *Controller) syncCheckStates(
 	wlInfo.update(wl, c.clock)
 	checksMap := slices.ToRefMap(wl.Status.AdmissionChecks, func(c *kueue.AdmissionCheckState) kueue.AdmissionCheckReference { return c.Name })
 	wlPatch := workload.BaseSSAWorkload(wl, true)
+	wlPatch.Status.RequeueState = wl.Status.RequeueState.DeepCopy()
 	recorderMessages := make([]string, 0, len(checkConfig))
 	updated := false
 	for check, prc := range checkConfig {
diff --git a/pkg/controller/core/localqueue_controller.go b/pkg/controller/core/localqueue_controller.go
index 75a6a75d3..2eb7c5e51 100644
--- a/pkg/controller/core/localqueue_controller.go
+++ b/pkg/controller/core/localqueue_controller.go
@@ -307,6 +307,7 @@ func (r *LocalQueueReconciler) reconcileConsumedUsage(ctx context.Context, lq *k
 func (r *LocalQueueReconciler) updateAdmissionFsStatus(ctx context.Context, lq *kueue.LocalQueue, consumedResources corev1.ResourceList) error {
 	lq.Status.FairSharing.AdmissionFairSharingStatus.ConsumedResources = consumedResources
 	lq.Status.FairSharing.AdmissionFairSharingStatus.LastUpdate = metav1.NewTime(r.clock.Now())
+	r.log.V(3).Info("Updated LocalQueue fair sharing status", "namespace", lq.Namespace, "name", lq.Name, "consumedResources", consumedResources)
 	return r.client.Status().Update(ctx, lq)
 }
 
diff --git a/pkg/controller/tas/node_failure_controller.go b/pkg/controller/tas/node_failure_controller.go
index 5ec8513a4..09db8ea29 100644
--- a/pkg/controller/tas/node_failure_controller.go
+++ b/pkg/controller/tas/node_failure_controller.go
@@ -87,7 +87,11 @@ func (r *nodeFailureReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 	readyCondition := utiltas.GetNodeCondition(&node, corev1.NodeReady)
 	if readyCondition.Status == corev1.ConditionTrue {
-		return ctrl.Result{}, nil
+		affectedWorkloads, err = r.getWorkloadsOnNode(ctx, req.Name)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		return ctrl.Result{}, r.removeNodeToReplaceAnnotation(ctx, req.Name, affectedWorkloads)
 	}
 	if features.Enabled(features.TASReplaceNodeOnPodTermination) {
 		return r.reconcileForReplaceNodeOnPodTermination(ctx, req.Name)
@@ -113,22 +117,16 @@ func (r *nodeFailureReconciler) Generic(event.TypedGenericEvent[*corev1.Node]) b
 }
 
 func (r *nodeFailureReconciler) Create(e event.TypedCreateEvent[*corev1.Node]) bool {
-	newNode := e.Object
-	if !utiltas.IsNodeStatusConditionTrue(newNode.Status.Conditions, corev1.NodeReady) {
-		r.log.V(4).Info("NodeReady is not true", "node", klog.KObj(newNode))
-		return true
-	}
-	r.log.V(5).Info("Node creation does not warrant reconcile for failure detection", "node", klog.KObj(newNode))
-	return false
+	return true
 }
 
 func (r *nodeFailureReconciler) Update(e event.TypedUpdateEvent[*corev1.Node]) bool {
-	newNode := e.ObjectNew
-	if !utiltas.IsNodeStatusConditionTrue(newNode.Status.Conditions, corev1.NodeReady) {
-		r.log.V(4).Info("NodeReady is not true", "node", klog.KObj(newNode))
+	newReady := utiltas.IsNodeStatusConditionTrue(e.ObjectNew.Status.Conditions, corev1.NodeReady)
+	oldReady := utiltas.IsNodeStatusConditionTrue(e.ObjectOld.Status.Conditions, corev1.NodeReady)
+	if oldReady != newReady {
+		r.log.V(4).Info("Node Ready status changed, triggering reconcile", "node", klog.KObj(e.ObjectNew), "oldReady", oldReady, "newReady", newReady)
 		return true
 	}
-	r.log.V(5).Info("Node update does not warrant reconcile for failure detection", "node", klog.KObj(newNode))
 	return false
 }
 
@@ -323,3 +321,44 @@ func (r *nodeFailureReconciler) reconcileForReplaceNodeOnPodTermination(ctx cont
 		return ctrl.Result{}, patchErr
 	}
 }
+
+// removeNodeToReplaceAnnotation finds workloads with the specified node in the NodeToReplaceAnnotation
+// and removes the annotation
+func (r *nodeFailureReconciler) removeNodeToReplaceAnnotation(ctx context.Context, nodeName string, affectedWorkloads sets.Set[types.NamespacedName]) error {
+	var workloadProcessingErrors []error
+	log := ctrl.LoggerFrom(ctx)
+	for wlKey := range affectedWorkloads {
+		log = log.WithValues("workload", wlKey)
+		// fetch workload.
+		var wl kueue.Workload
+		if err := r.client.Get(ctx, wlKey, &wl); err != nil {
+			if apierrors.IsNotFound(err) {
+				log.V(4).Info("Workload not found, skipping")
+			} else {
+				log.Error(err, "Failed to get workload")
+				workloadProcessingErrors = append(workloadProcessingErrors, err)
+			}
+			continue
+		}
+
+		if wl.Annotations != nil && wl.Annotations[kueuealpha.NodeToReplaceAnnotation] != nodeName {
+			continue
+		}
+
+		err := clientutil.Patch(ctx, r.client, &wl, true, func() (bool, error) {
+			log.V(4).Info(fmt.Sprintf("Removing node from %s annotation", kueuealpha.NodeToReplaceAnnotation), "failedNode", nodeName)
+			delete(wl.Annotations, kueuealpha.NodeToReplaceAnnotation)
+			return true, nil
+		})
+		if err != nil {
+			log.Error(err, "Failed to patch workload annotation")
+			workloadProcessingErrors = append(workloadProcessingErrors, err)
+			continue
+		}
+		log.V(3).Info("Successfully cleared the NodeToReplace annotation")
+	}
+	if len(workloadProcessingErrors) > 0 {
+		return errors.Join(workloadProcessingErrors...)
+	}
+	return nil
+}
diff --git a/pkg/scheduler/preemption/classical/candidate_generator.go b/pkg/scheduler/preemption/classical/candidate_generator.go
index 11ce99e4e..0e6a4fe66 100644
--- a/pkg/scheduler/preemption/classical/candidate_generator.go
+++ b/pkg/scheduler/preemption/classical/candidate_generator.go
@@ -20,6 +20,7 @@ import (
 	"sort"
 	"time"
 
+	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/utils/clock"
@@ -73,17 +74,17 @@ func splitEvicted(workloads []*candidateElem) ([]*candidateElem, []*candidateEle
 // with and without borrowing. The runs are independent which means that the same candidates
 // might be returned for both, but note that the candidates with borrrowing are a subset of
 // candidates without borrowing.
-func NewCandidateIterator(hierarchicalReclaimCtx *HierarchicalPreemptionCtx, frsNeedPreemption sets.Set[resources.FlavorResource], snapshot *cache.Snapshot, clock clock.Clock, ordering func(*workload.Info, *workload.Info, kueue.ClusterQueueReference, time.Time) bool) *candidateIterator {
+func NewCandidateIterator(hierarchicalReclaimCtx *HierarchicalPreemptionCtx, frsNeedPreemption sets.Set[resources.FlavorResource], snapshot *cache.Snapshot, clock clock.Clock, ordering func(logr.Logger, *workload.Info, *workload.Info, kueue.ClusterQueueReference, time.Time) bool) *candidateIterator {
 	sameQueueCandidates := collectSameQueueCandidates(hierarchicalReclaimCtx)
 	hierarchyCandidates, priorityCandidates := collectCandidatesForHierarchicalReclaim(hierarchicalReclaimCtx)
 	sort.Slice(sameQueueCandidates, func(i, j int) bool {
-		return ordering(sameQueueCandidates[i].wl, sameQueueCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
+		return ordering(hierarchicalReclaimCtx.Log, sameQueueCandidates[i].wl, sameQueueCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
 	})
 	sort.Slice(priorityCandidates, func(i, j int) bool {
-		return ordering(priorityCandidates[i].wl, priorityCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
+		return ordering(hierarchicalReclaimCtx.Log, priorityCandidates[i].wl, priorityCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
 	})
 	sort.Slice(hierarchyCandidates, func(i, j int) bool {
-		return ordering(hierarchyCandidates[i].wl, hierarchyCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
+		return ordering(hierarchicalReclaimCtx.Log, hierarchyCandidates[i].wl, hierarchyCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
 	})
 
 	evictedHierarchicalReclaimCandidates, nonEvictedHierarchicalReclaimCandidates := splitEvicted(hierarchyCandidates)
diff --git a/pkg/scheduler/preemption/classical/hierarchical_preemption.go b/pkg/scheduler/preemption/classical/hierarchical_preemption.go
index ca3159146..5f8fb87d8 100644
--- a/pkg/scheduler/preemption/classical/hierarchical_preemption.go
+++ b/pkg/scheduler/preemption/classical/hierarchical_preemption.go
@@ -17,6 +17,7 @@ limitations under the License.
 package classical
 
 import (
+	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
@@ -58,6 +59,7 @@ func (m preemptionVariant) PreemptionReason() string {
 }
 
 type HierarchicalPreemptionCtx struct {
+	Log               logr.Logger
 	Wl                *kueue.Workload
 	Cq                *cache.ClusterQueueSnapshot
 	FrsNeedPreemption sets.Set[resources.FlavorResource]
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 5a3eee051..978dd9d01 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -208,6 +208,7 @@ type preemptionAttemptOpts struct {
 // fits
 func (p *Preemptor) classicalPreemptions(preemptionCtx *preemptionCtx) []*Target {
 	hierarchicalReclaimCtx := &classical.HierarchicalPreemptionCtx{
+		Log:               preemptionCtx.log,
 		Wl:                preemptionCtx.preemptor.Obj,
 		Cq:                preemptionCtx.preemptorCQ,
 		FrsNeedPreemption: preemptionCtx.frsNeedPreemption,
@@ -380,7 +381,7 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, strategies []f
 		return nil
 	}
 	sort.Slice(candidates, func(i, j int) bool {
-		return CandidatesOrdering(candidates[i], candidates[j], preemptionCtx.preemptorCQ.Name, p.clock.Now())
+		return CandidatesOrdering(preemptionCtx.log, candidates[i], candidates[j], preemptionCtx.preemptorCQ.Name, p.clock.Now())
 	})
 	if logV := preemptionCtx.log.V(5); logV.Enabled() {
 		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
@@ -542,7 +543,7 @@ func resourceUsagePreemptionEnabled(a, b *workload.Info) bool {
 // 2. (AdmissionFairSharing only) Workloads with lower LocalQueue's usage first
 // 3. Workloads with lower priority first.
 // 4. Workloads admitted more recently first.
-func CandidatesOrdering(a, b *workload.Info, cq kueue.ClusterQueueReference, now time.Time) bool {
+func CandidatesOrdering(log logr.Logger, a, b *workload.Info, cq kueue.ClusterQueueReference, now time.Time) bool {
 	aEvicted := meta.IsStatusConditionTrue(a.Obj.Status.Conditions, kueue.WorkloadEvicted)
 	bEvicted := meta.IsStatusConditionTrue(b.Obj.Status.Conditions, kueue.WorkloadEvicted)
 	if aEvicted != bEvicted {
@@ -556,6 +557,9 @@ func CandidatesOrdering(a, b *workload.Info, cq kueue.ClusterQueueReference, now
 
 	if features.Enabled(features.AdmissionFairSharing) && resourceUsagePreemptionEnabled(a, b) {
 		if a.LocalQueueFSUsage != b.LocalQueueFSUsage {
+			log.V(3).Info("Comparing workloads by LocalQueue fair sharing usage",
+				"workloadA", klog.KObj(a.Obj), "queueA", a.Obj.Spec.QueueName, "usageA", a.LocalQueueFSUsage,
+				"workloadB", klog.KObj(b.Obj), "queueB", b.Obj.Spec.QueueName, "usageB", b.LocalQueueFSUsage)
 			return *a.LocalQueueFSUsage > *b.LocalQueueFSUsage
 		}
 	}
diff --git a/site/content/en/docs/concepts/multikueue.md b/site/content/en/docs/concepts/multikueue.md
index 6ab9505c3..9b024cdad 100644
--- a/site/content/en/docs/concepts/multikueue.md
+++ b/site/content/en/docs/concepts/multikueue.md
@@ -50,7 +50,9 @@ To enable multi-cluster dispatching, you need to assign a Job to a ClusterQueue
 The dispatching flow works as follows:
 
 1. When the Job's Workload obtains a `QuotaReservation` in the manager cluster,
-   a copy of that Workload is created in all configured worker clusters.
+   the dispatcher determines the worker clusters where the Workload should be created.
+   Depending on the configured dispatching algorithm (e.g., AllAtOnce, Incremental, or a custom approach),
+   the Workload may be created in all configured worker clusters or in a subset of them.
 2. When a worker cluster admits one of these remote Workloads:
    - The manager deletes the Workloads from the other clusters.
    - The manager creates a copy of the Job in the selected worker cluster and labels it
@@ -61,6 +63,48 @@ The dispatching flow works as follows:
    - The manager performs a final status sync.
    - It then deletes the corresponding objects from the worker cluster.
 
+## Workload Dispatching
+
+{{% alert title="Note" color="primary" %}}
+The MultiKueue Dispatcher mechanism is available since Kueue v0.13.
+{{% /alert %}}
+Provides a flexible way to control how workloads are distributed across worker clusters
+by allowing users to choose between built-in dispatching algorithms or implement custom ones.
+This mechanism ensures efficient workload placement while minimizing resource contention and preemptions across clusters.
+
+The `status.nominatedClusterNames` field lists the worker clusters currently being considered for scheduling the Workload,
+as determined by the dispatching algorithm, and is updated while the Workload is pending admission.
+
+The `status.clusterName` field specifies the worker cluster where the Workload has been successfully admitted.
+Once the field is set, it becomes immutable and the `status.nominatedClusterNames` field is reset,
+and it is no longer possible to set it. 
+
+This ensures that the Workload's cluster assignment is finalized and prevents further nomination of clusters.
+
+### AllAtOnce (Default Mode):
+In this mode, the Workload is copied to all available worker clusters as soon as it obtains a QuotaReservation in the manager cluster.
+This approach ensures the fastest possible admission by allowing all clusters to compete for the Workload simultaneously.
+
+### Incremental:
+This mode introduces a gradual dispatching strategy where clusters are nominated in rounds.
+Initially, all worker clusters are sorted in dictionary order, and a subset of up to 3 clusters is selected from the sorted list.
+The Workload is copied only to these nominated clusters.
+If none of the nominated clusters admit the Workload within a fixed duration (5 minutes),
+an additional up to 3 clusters are incrementally added in subsequent rounds, following the same dictionary order.
+
+### External (Custom implementation):
+In this mode, the selection of worker clusters is delegated to an external controller.
+The external controller is responsible for setting the `.status.nominatedClusterNames` field in the Workload to specify the clusters where it should be copied.
+
+The MultiKueue Workload Controller synchronizes the Workload with the nominated clusters.
+
+Known Limitation:
+{{% alert title="Warning" color="primary" %}}
+For the external controller to patch the `.status.nominatedClusterNames` field, it must use the kueue-admission field manager.
+This is required because the kueue-admission field manager is responsible for managing updates to the `.status.nominatedClusterNames` field.
+Without this, the Kueue is not able to admit the MultiKueue workloads.
+{{% /alert %}}
+
 ## Supported Job Types
 
 MultiKueue supports a wide variety of workloads. You can learn how to:
diff --git a/site/content/en/docs/concepts/topology_aware_scheduling.md b/site/content/en/docs/concepts/topology_aware_scheduling.md
index 7943b210f..1d3814254 100644
--- a/site/content/en/docs/concepts/topology_aware_scheduling.md
+++ b/site/content/en/docs/concepts/topology_aware_scheduling.md
@@ -85,10 +85,15 @@ following annotations set at the PodTemplate level:
 	then the next topology level up is considered. If the PodSet cannot fit
 	at the highest topology level, then it gets admitted as distributed
 	among multiple topology domains.
-- `kueue.x-k8s.io/podset-required-topology` - indicates indicates that a PodSet
+- `kueue.x-k8s.io/podset-required-topology` - indicates that a PodSet
   requires Topology Aware Scheduling, and requires scheduling all pods on nodes
 	within the same topology domain corresponding to the topology level
 	indicated by the annotation value (e.g. within a rack or within a block).
+- `kueue.x-k8s.io/podset-unconstrained-topology` - indicates that a PodSet requires
+    Topology Aware Scheduling, and requires scheduling all pods on any nodes without
+    topology considerations. In other words, this considers if all pods could be accommodated 
+    within any nodes which helps to minimize fragmentation by filling the small gaps
+    on nodes across the cluster.
 
 #### Example
 
diff --git a/site/content/zh-CN/docs/contribution_guidelines/testing.md b/site/content/zh-CN/docs/contribution_guidelines/testing.md
index d8b080816..240e820a8 100644
--- a/site/content/zh-CN/docs/contribution_guidelines/testing.md
+++ b/site/content/zh-CN/docs/contribution_guidelines/testing.md
@@ -1,59 +1,59 @@
 ---
-title: "Running and debugging tests"
-linkTitle: "Running tests"
+title: "运行和调试测试"
+linkTitle: "运行测试"
 weight: 25
 description: >
-  Running and debugging tests
+  运行和调试测试
 ---
 
-## Running presubmission verification tests
+## 运行预提交验证测试 {#running-presubmission-verification-tests}
 ```shell
 make verify
 ```
 
-## Running unit tests
-To run all unit tests:
+## 运行单元测试 {#running-unit-tests}
+运行所有单元测试：
 ```shell
 make test
 ```
 
-To run unit tests for webhooks:
+运行 Webhook 的单元测试：
 ```shell
 go test ./pkg/webhooks
 ```
-To run tests that match `TestValidateClusterQueue` regular expression [ref](https://pkg.go.dev/cmd/go#hdr-Testing_flags):
+[参考文档](https://pkg.go.dev/cmd/go#hdr-Testing_flags)运行与 `TestValidateClusterQueue` 正则表达式匹配的测试：
 ```shell
 go test ./pkg/webhooks -run TestValidateClusterQueue
 ```
 
-### Running unit tests with race detection
+### 使用竞态检测运行单元测试 {#running-unit-tests-with-race-detection}
 
-Use `-race` to enable Go's built-in race detector:
+使用 `-race` 启用 Go 内置的竞态检测器：
 ```shell
 go test ./pkg/scheduler/preemption/ -race
 ```
 
-### Running unit tests with stress
+### 使用压力测试运行单元测试 {#running-unit-tests-with-stress}
 
-To run unit tests in a loop and collect failures:
+循环运行单元测试并收集失败情况：
 ```shell
-# install go stress
+# 安装 go stress
 go install golang.org/x/tools/cmd/stress@latest
-# compile tests (you can add -race for race detection)
+# 编译测试（可以添加 -race 进行竞态检测）
 go test ./pkg/scheduler/preemption/ -c
-# it loops and reports failures
+# 循环运行并报告失败
 stress ./preemption.test -test.run TestPreemption
 ```
 
-## Running integration tests
+## 运行集成测试 {#running-integration-tests}
 
 ```shell
 make test-integration
 ```
 
-For running a subset of tests, see [Running subset of tests](#running-subset-of-integration-or-e2e-tests).
+关于运行测试子集，请参阅[运行测试子集](#running-subset-of-integration-or-e2e-tests)。
 
-## Running e2e tests using custom build
+## 使用自定义构建运行 e2e 测试 {#running-e2e-tests-using-custom-build}
 ```shell
 make kind-image-build
 make test-e2e
@@ -64,37 +64,37 @@ make test-e2e-kueueviz
 make test-multikueue-e2e
 ```
 
-You can specify the Kubernetes version used for running the e2e tests by setting the `E2E_K8S_FULL_VERSION` variable:
+你可以通过设置 `E2E_K8S_FULL_VERSION` 变量来指定用于运行 e2e 测试的 Kubernetes 版本：
 ```shell
 E2E_K8S_FULL_VERSION=1.33.1 make test-e2e
 ```
 
-For running a subset of tests, see [Running subset of tests](#running-subset-of-integration-or-e2e-tests).
+关于运行测试子集，请参阅 [运行测试子集](#running-subset-of-integration-or-e2e-tests)。
 
-## Increase logging verbosity
-You can change log level (for example, set -5 to increase verbosity) using `TEST_LOG_LEVEL` variables.
-By default, `TEST_LOG_LEVEL=-3`.
+## 增加日志详细程度 {#increase-logging-verbosity}
+你可以使用 `TEST_LOG_LEVEL` 变量更改日志级别（例如，设置 -5 来增加详细程度）。
+默认情况下，`TEST_LOG_LEVEL=-3`。
 
-## Debug tests in VSCode
-It is possible to debug unit and integration tests in VSCode.
-You need to have the [Go extension](https://marketplace.visualstudio.com/items?itemName=golang.Go) installed.
-Now you will have `run test | debug test` text buttons above lines like
+## 在 VSCode 中调试测试 {#debug-tests-in-vscode}
+可以在 VSCode 中调试单元测试和集成测试。
+你需要安装 [Go 扩展](https://marketplace.visualstudio.com/items?itemName=golang.Go)。
+现在你将在如下行上方看到 `run test | debug test` 文本按钮：
 ```go
 func TestValidateClusterQueue(t *testing.T) {
 ```
-You can click on the `debug test` to debug a specific test.
+你可以点击 `debug test` 来调试特定测试。
 
-For integration tests, an additional step is needed.  In settings.json, you need to add two variables inside `go.testEnvVars`:
-- Run `ENVTEST_K8S_VERSION=1.33 make envtest && ./bin/setup-envtest use $ENVTEST_K8S_VERSION -p path` and assign the path to the `KUBEBUILDER_ASSETS` variable
-- Set `KUEUE_BIN` to the `bin` directory within your cloned Kueue repository
+对于集成测试，需要额外的步骤。在 settings.json 中，你需要在 `go.testEnvVars` 内添加两个变量：
+- 运行 `ENVTEST_K8S_VERSION=1.33 make envtest && ./bin/setup-envtest use $ENVTEST_K8S_VERSION -p path` 并将路径分配给 `KUBEBUILDER_ASSETS` 变量
+- 将 `KUEUE_BIN` 设置为你的 Kueue 仓库克隆目录内的 `bin` 目录
 ```json
 "go.testEnvVars": {
-    "KUBEBUILDER_ASSETS": "<path from output above>",
-    "KUEUE_BIN": "<path-to-your-kueue-folder>/bin",
+    "KUBEBUILDER_ASSETS": "<上面输出的路径>",
+    "KUEUE_BIN": "<你的-kueue-文件夹路径>/bin",
   },
 ```
 
-For e2e tests, you can also use [Ginkgo Test Explorer](https://marketplace.visualstudio.com/items?itemName=joselitofilho.ginkgotestexplorer).  You need to add the following variables to settings.json:
+对于 e2e 测试，你也可以使用 [Ginkgo Test Explorer](https://marketplace.visualstudio.com/items?itemName=joselitofilho.ginkgotestexplorer)。你需要在 settings.json 中添加以下变量：
 ```json
  "ginkgotestexplorer.testEnvVars": {
         "KIND_CLUSTER_NAME": "kind",
@@ -104,79 +104,79 @@ For e2e tests, you can also use [Ginkgo Test Explorer](https://marketplace.visua
         "KIND": "<your_kueue_path>/bin/kind",
     },
 ```
-and then you can use GUI of the Ginkgo Test Explorer to run individual tests, provided you started kind clanter (see [here](#attaching-e2e-tests-to-an-existing-kind-cluster) for the instructions).
+然后你可以使用 Ginkgo Test Explorer 的 GUI 来运行单个测试，前提是你已经启动了 kind 集群（有关说明，请参阅[这里](#attaching-e2e-tests-to-an-existing-kind-cluster)）。
 
-## Attaching e2e tests to an existing kind cluster
-You can use the following approach to start up a kind cluster and then run e2e tests from commandline or VSCode,
-attaching them to the existing cluster. For example, suppose you want to test some of the multikueue-e2e tests.
+## 将 e2e 测试附加到现有的 kind 集群 {#attaching-e2e-tests-to-an-existing-kind-cluster}
+你可以使用以下方法来启动 kind 集群，然后从命令行或 VSCode 运行 e2e 测试，
+将它们附加到现有集群。例如，假设你想测试一些 multikueue-e2e 测试。
 
-Run `E2E_RUN_ONLY_ENV=true make kind-image-build test-multikueue-e2e` and wait for the `Press Enter to cleanup.` to appear.
+运行 `E2E_RUN_ONLY_ENV=true make kind-image-build test-multikueue-e2e` 并等待 `Press Enter to cleanup.` 出现。
 
-The cluster is ready, and now you can run tests from another terminal:
+集群已准备就绪，现在你可以从另一个终端运行测试：
 ```shell
 <your_kueue_path>/bin/ginkgo --json-report ./ginkgo.report -focus "MultiKueue when Creating a multikueue admission check Should run a jobSet on worker if admitted" -r
 ```
-or from VSCode.
+或从 VSCode 运行。
 
-## Running subset of integration or e2e tests
-### Use Ginkgo --focus arg
+## 运行集成或 e2e 测试子集 {#running-subset-of-integration-or-e2e-tests}
+### 使用 Ginkgo --focus 参数 {#use-ginkgo-focus-arg}
 ```shell
 GINKGO_ARGS="--focus=Scheduler" make test-integration
 GINKGO_ARGS="--focus=Creating a Pod requesting TAS" make test-e2e
 ```
-### Use ginkgo.FIt
-If you want to focus on specific tests, you can change
-`ginkgo.It` to `ginkgo.FIt` for these tests.
-For more details, see [here](https://onsi.github.io/ginkgo/#focused-specs).
-Then the other tests will be skipped.
-For example, you can change
+### 使用 ginkgo.FIt {#use-ginkgo-fit}
+如果你想专注于特定测试，可以将这些测试的
+`ginkgo.It` 更改为 `ginkgo.FIt`。
+有关更多详细信息，请参阅[这里](https://onsi.github.io/ginkgo/#focused-specs)。
+然后其他测试将被跳过。
+例如，你可以将
 ```go
 ginkgo.It("Should place pods based on the ranks-ordering", func() {
 ```
-to
+更改为
 ```go
 ginkgo.FIt("Should place pods based on the ranks-ordering", func() {
 ```
-and then run
+然后运行
 ```shell
-# build and pull image
+# 构建并拉取镜像
 make test-tas-e2e
 ```
-to test a particular TAS e2e test.
+来测试特定的 TAS e2e 测试。
 
-### Use INTEGRATION_TARGET
+### 使用 INTEGRATION_TARGET {#use-integration-target}
 ```shell
 INTEGRATION_TARGET='test/integration/singlecluster/scheduler' make test-integration
 ```
 
-## Flaky integration/e2e tests
-You can use --until-it-fails or --repeat=N arguments to Ginkgo to run tests repeatedly, such as:
+## 不稳定的集成/e2e 测试 {#flaky-integration-e2e-tests}
+你可以使用 --until-it-fails 或 --repeat=N 参数来让 Ginkgo 重复运行测试，例如：
 ```shell
 GINKGO_ARGS="--until-it-fails" make test-integration
 GINKGO_ARGS="--repeat=10" make test-e2e
 ```
-See more [here](https://onsi.github.io/ginkgo/#repeating-spec-runs-and-managing-flaky-specs)
+更多信息请参阅[这里](https://onsi.github.io/ginkgo/#repeating-spec-runs-and-managing-flaky-specs)
 
-### Adding stress
-You can run [stress](https://github.com/resurrecting-open-source-projects/stress) tool to increase CPU load during tests.  For example, if you're on Debian-based Linux:
+### 添加压力 {#adding-stress}
+你可以运行 [stress](https://github.com/resurrecting-open-source-projects/stress) 工具来在测试期间增加 CPU 负载。例如，如果你在基于 Debian 的 Linux 上：
 ```shell
-# install stress:
+# 安装 stress：
 sudo apt install stress
-# run stress alongside tests
+# 与测试一起运行 stress
 /usr/bin/stress --cpu 80
 ```
 
-### Analyzing logs
-Kueue runs as a regular pod on a worker node, and in e2e tests there are 2 replicas running.  The Kueue logs are located in `kind-worker/pods/kueue-system_kueue-controller-manager*/manager` and `kind-worker2/pods/kueue-system_kueue-controller-manager*/manager` folders.
+### 分析日志 {#analyzing-logs}
+Kueue 作为常规 pod 在 worker 节点上运行，在 e2e 测试中有 2 个副本在运行。Kueue 日志位于 `kind-worker/pods/kueue-system_kueue-controller-manager*/manager` 和 `kind-worker2/pods/kueue-system_kueue-controller-manager*/manager` 文件夹中。
 
-For each log message you can from which file and line the message is coming from:
+对于每条日志消息，你可以看到消息来自哪个文件和行：
 ```log
 2025-02-03T15:51:51.502425029Z stderr F 2025-02-03T15:51:51.502117824Z	LEVEL(-2)	cluster-queue-reconciler	core/clusterqueue_controller.go:341	ClusterQueue update event	{"clusterQueue": {"name":"cluster-queue"}}
 ```
-Here, it's `core/clusterqueue_controller.go:341`.
+这里，它是 `core/clusterqueue_controller.go:341`。
 
-### See also
-- [Kubernetes testing guide](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/testing.md)
-- [Integration Testing in Kubernetes](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/integration-tests.md)
-- [End-to-End Testing in Kubernetes](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/e2e-tests.md)
-- [Flaky Tests in Kubernetes](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/flaky-tests.md)
+### 另请参阅 {#see-also}
+- [Kubernetes 测试指南](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/testing.md)
+- [Kubernetes 中的集成测试](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/integration-tests.md)
+- [Kubernetes 中的端到端测试](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/e2e-tests.md)
+- [Kubernetes 中的不稳定测试](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/flaky-tests.md)
diff --git a/site/content/zh-CN/docs/tasks/run/multikueue/jobsets.md b/site/content/zh-CN/docs/tasks/run/multikueue/jobsets.md
index 7177cc639..6ad7d2bb1 100644
--- a/site/content/zh-CN/docs/tasks/run/multikueue/jobsets.md
+++ b/site/content/zh-CN/docs/tasks/run/multikueue/jobsets.md
@@ -1,22 +1,20 @@
 ---
-title: "Run Jobsets in Multi-Cluster"
+title: "在多集群环境中运行 Jobset"
 linkTitle: "Jobsets"
 weight: 3
 date: 2024-09-25
-description: >
-  Run a MultiKueue scheduled JobSet.
+description: 运行 MultiKueue 调度的 Jobset。
 ---
 
-## Before you begin
+## 开始之前 {#before-you-begin}
 
-Check the [MultiKueue installation guide](/docs/tasks/manage/setup_multikueue) on how to properly setup MultiKueue clusters.
+请查阅 [MultiKueue 安装指南](/zh-CN/docs/tasks/manage/setup_multikueue)了解如何正确设置 MultiKueue 集群。
 
-For the ease of setup and use we recommend using at least Kueue v0.8.1 and JobSet v0.6.0 (see [JobSet Installation](https://jobset.sigs.k8s.io/docs/installation/) for more details).
+为方便安装和使用，建议至少使用 Kueue v0.8.1 和 JobSet v0.6.0（更多详情见 [JobSet 安装文档](https://jobset.sigs.k8s.io/docs/installation/)）。
 
-## Run Jobsets example
+## 运行 Jobset 示例 {#run-jobsets-example}
 
-Once the setup is complete you can test it by running the [`jobset-sample.yaml`](/docs/tasks/run/jobsets/#example-jobset) Job. 
+完成设置后，你可以通过运行 [`jobset-sample.yaml`](/zh-CN/docs/tasks/run/jobsets/#example-jobset) Job 进行测试。
 
-In that setup, the `spec.managedBy` field is defaulted to `kueue.x-k8s.io/multikueue`
-automatically, if not specified, as long as  the `kueue.x-k8s.io/queue-name` annotation
-is specified and the corresponding Cluster Queue uses the Multi Kueue admission check.
\ No newline at end of file
+在该设置中，如果未显式指定 spec.managedBy 字段的取值，则取值自动默认为 `kueue.x-k8s.io/multikueue`。
+前提是已设置了 `kueue.x-k8s.io/queue-name` 注解，并且对应的 ClusterQueue 使用了 MultiKueue 的准入检查。
\ No newline at end of file
diff --git a/site/content/zh-CN/docs/tasks/run/multikueue/kubeflow.md b/site/content/zh-CN/docs/tasks/run/multikueue/kubeflow.md
index 1d7b18f46..49bceaf0e 100644
--- a/site/content/zh-CN/docs/tasks/run/multikueue/kubeflow.md
+++ b/site/content/zh-CN/docs/tasks/run/multikueue/kubeflow.md
@@ -1,44 +1,43 @@
 ---
-title: "Run Kubeflow Jobs in Multi-Cluster"
+title: "在多集群环境中运行 Kubeflow Job"
 linkTitle: "Kubeflow"
 weight: 4
 date: 2024-09-25
-description: >
-  Run a MultiKueue scheduled Kubeflow Jobs.
+description: 运行 MultiKueue 调度的 Kubeflow Job。
 ---
 
-## Before you begin
+## 开始之前 {#before-you-begin}
 
-Check the [MultiKueue installation guide](/docs/tasks/manage/setup_multikueue) on how to properly setup MultiKueue clusters.
+请查阅 [MultiKueue 安装指南](/zh-CN/docs/tasks/manage/setup_multikueue)了解如何正确设置 MultiKueue 集群。
 
-For the ease of setup and use we recommend using at least Kueue v0.11.0 and for Kubeflow Trainer at least v1.9.0.
+为方便安装和使用，建议使用 Kueue v0.11.0 和 Kubeflow Trainer v1.9.0 以上版本。
 
-See [Trainer Installation](https://www.kubeflow.org/docs/components/training/installation/#installing-the-training-operator) for installation and configuration details of Trainer.
+有关 Trainer 的安装和配置详情，请参见 [Trainer 安装文档](https://www.kubeflow.org/docs/components/training/installation/#installing-the-training-operator)。
 
-{{% alert title="Note" color="primary" %}}
-Before the [ManagedBy feature](https://github.com/kubeflow/trainer/issues/2193) was supported in Kueue (below v0.11.0), the installation of Kubeflow Trainer in the <b>Manager Cluster</b> must be limited to CRDs only.
+{{% alert title="注意" color="primary" %}}
+在 Kueue（低于 v0.11.0）支持 [ManagedBy 特性](https://github.com/kubeflow/trainer/issues/2193)之前，<b>管理集群</b>上的 Kubeflow Trainer 安装必须仅限于 CRD。
 
-To install the CRDs run:
+要安装 CRD，请运行：
 ```bash
 kubectl apply -k "github.com/kubeflow/trainer.git/manifests/base/crds?ref=v1.9.0"
 ```
 {{% /alert %}}
 
-## MultiKueue integration
+## MultiKueue 集成 {#multikueue-integration}
 
-Once the setup is complete you can test it by running one of the Kubeflow Jobs e.g. PyTorchJob [`sample-pytorchjob.yaml`](/docs/tasks/run/kubeflow/pytorchjobs/#sample-pytorchjob). 
+完成设置后，你可以通过运行其中一个 Kubeflow Job（如 PyTorchJob [`sample-pytorchjob.yaml`](/zh-CN/docs/tasks/run/kubeflow/pytorchjobs/#sample-pytorchjob)）进行测试。
 
-{{% alert title="Note" color="primary" %}}
-Note: Kueue defaults the `spec.runPolicy.managedBy` field to `kueue.x-k8s.io/multikueue` on the management cluster for all Kubeflow Jobs. 
+{{% alert title="注意" color="primary" %}}
+注意：Kueue 会在管理集群上的所有 Kubeflow Job 默认设置 `spec.runPolicy.managedBy` 字段为 `kueue.x-k8s.io/multikueue`。
 
-This allows the Trainer to ignore the Jobs managed by MultiKueue on the management cluster, and in particular skip Pod creation. 
+这使得 Trainer 能够忽略由 MultiKueue 管理的 Job，特别是跳过 Pod 的创建。
 
-The pods are created and the actual computation will happen on the mirror copy of the Job on the selected worker cluster. 
-The mirror copy of the Job does not have the field set.
+Pod 会在选定的工作集群上的 Job 镜像副本中被创建并实际运行。
+Job 镜像副本未设置此字段。
 {{% /alert %}}
 
-## Working alongside MPI Operator
-In order for MPI-operator and Trainer to work on the same cluster it is required that:
-1. `kubeflow.org_mpijobs.yaml` entry is removed from `base/crds/kustomization.yaml` - https://github.com/kubeflow/trainer/issues/1930
-2. Trainer deployment is modified to enable all kubeflow jobs except for MPI -  https://github.com/kubeflow/trainer/issues/1777
+## 与 MPI Operator 协同工作 {#working-alongside-mpi-operator}
+为了让 MPI-operator 和 Trainer 能在同一集群上工作，需要：
+1. 从 `base/crds/kustomization.yaml` 中移除 `kubeflow.org_mpijobs.yaml` - https://github.com/kubeflow/trainer/issues/1930
+2. 修改 Trainer 部署以启用除 MPI 以外的所有 Kubeflow Job -  https://github.com/kubeflow/trainer/issues/1777
   
\ No newline at end of file


Test patch:
diff --git a/pkg/controller/tas/node_failure_controller_test.go b/pkg/controller/tas/node_failure_controller_test.go
index d9dd43a93..b012fa5ba 100644
--- a/pkg/controller/tas/node_failure_controller_test.go
+++ b/pkg/controller/tas/node_failure_controller_test.go
@@ -84,6 +84,11 @@ func TestNodeFailureReconciler(t *testing.T) {
 		Admitted(true).
 		Obj()
 
+	workloadWithAnnotation := baseWorkload.DeepCopy()
+	workloadWithAnnotation.Annotations = map[string]string{
+		kueuealpha.NodeToReplaceAnnotation: nodeName,
+	}
+
 	workloadWithTwoNodes := utiltesting.MakeWorkload(wlName, nsName).
 		Finalizers(kueue.ResourceInUseFinalizerName).
 		PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 2).Request(corev1.ResourceCPU, "1").Obj()).
@@ -148,6 +153,16 @@ func TestNodeFailureReconciler(t *testing.T) {
 			reconcileRequests: []reconcile.Request{{NamespacedName: types.NamespacedName{Name: nodeName}}},
 			wantFailedNode:    "",
 		},
+		"Node becomes healthy, annotation is removed": {
+			initObjs: []client.Object{
+				newNodeTest(nodeName, corev1.ConditionTrue, fakeClock, time.Duration(0)),
+				workloadWithAnnotation.DeepCopy(),
+				basePod.DeepCopy(),
+			},
+			reconcileRequests: []reconcile.Request{{NamespacedName: types.NamespacedName{Name: nodeName}}},
+			wantFailedNode:    "",
+		},
+
 		"Node Found and Unhealthy (NotReady), delay not passed - not marked as unavailable": {
 			initObjs: []client.Object{
 				newNodeTest(nodeName, corev1.ConditionFalse, fakeClock, time.Duration(0)),
diff --git a/pkg/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index 94583eed8..66dc406f5 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -2894,12 +2894,13 @@ func TestCandidatesOrdering(t *testing.T) {
 			enableFeatureGates: []featuregate.Feature{features.AdmissionFairSharing},
 		}}
 
+	_, log := utiltesting.ContextWithLog(t)
 	for _, tc := range cases {
 		for _, gate := range tc.enableFeatureGates {
 			features.SetFeatureGateDuringTest(t, gate, true)
 		}
 		sort.Slice(tc.candidates, func(i int, j int) bool {
-			return CandidatesOrdering(&tc.candidates[i], &tc.candidates[j], kueue.ClusterQueueReference(preemptorCq), now)
+			return CandidatesOrdering(log, &tc.candidates[i], &tc.candidates[j], kueue.ClusterQueueReference(preemptorCq), now)
 		})
 		got := slices.Map(tc.candidates, func(c *workload.Info) workload.Reference {
 			return workload.Reference(c.Obj.Name)
diff --git a/test/e2e/certmanager/metrics_test.go b/test/e2e/certmanager/metrics_test.go
index 45049f60a..29e674753 100644
--- a/test/e2e/certmanager/metrics_test.go
+++ b/test/e2e/certmanager/metrics_test.go
@@ -36,7 +36,7 @@ const (
 	serviceAccountName           = "kueue-controller-manager"
 	metricsReaderClusterRoleName = "kueue-metrics-reader"
 	metricsServiceName           = "kueue-controller-manager-metrics-service"
-	certSecretName               = "metrics-server-cert"
+	certSecretName               = "kueue-metrics-server-cert"
 	certMountPath                = "/etc/kueue/metrics/certs"
 )
 
@@ -101,7 +101,7 @@ var _ = ginkgo.Describe("Metrics", ginkgo.Ordered, func() {
 		}
 		util.MustCreate(ctx, k8sClient, curlPod)
 
-		ginkgo.By("Waiting for metrics-server-cert secret", func() {
+		ginkgo.By("Waiting for kueue-metrics-server-cert secret", func() {
 			gomega.Eventually(func(g gomega.Gomega) {
 				secret := &corev1.Secret{}
 				g.Expect(k8sClient.Get(ctx, client.ObjectKey{
diff --git a/test/e2e/config/certmanager/dev/kustomization.yaml b/test/e2e/config/certmanager/dev/kustomization.yaml
new file mode 100644
index 000000000..063319fea
--- /dev/null
+++ b/test/e2e/config/certmanager/dev/kustomization.yaml
@@ -0,0 +1,7 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+namePrefix: kueue-
+
+resources:
+- ../../../../../config/components/certmanager
diff --git a/test/e2e/config/certmanager/kustomization.yaml b/test/e2e/config/certmanager/kustomization.yaml
index 3f0ab8c0e..66a3060ac 100644
--- a/test/e2e/config/certmanager/kustomization.yaml
+++ b/test/e2e/config/certmanager/kustomization.yaml
@@ -4,7 +4,7 @@ kind: Kustomization
 namespace: kueue-system
 
 resources:
-- ../../../../config/components/certmanager
+- ./dev
 - ../../../../config/dev
 
 transformers:
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index 38801378f..371e36edf 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package mke2e
 
 import (
+	"context"
 	"fmt"
 	"os/exec"
 	"regexp"
@@ -497,14 +498,8 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the job is completed", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					workerWl := &kueue.Workload{}
-					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					workerJob := &batchv1.Job{}
-					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(job), workerJob)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(job), workerJob)).To(utiltesting.BeNotFoundError())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				expectObjectToBeDeletedOnWorkerClusters(ctx, createdLeaderWorkload)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, job)
 
 				createdJob := &batchv1.Job{}
 				gomega.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
@@ -541,24 +536,8 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 
 			createdLeaderWorkload := &kueue.Workload{}
 			wlLookupKey := types.NamespacedName{Name: workloadjobset.GetWorkloadNameForJobSet(jobSet.Name, jobSet.UID), Namespace: managerNs.Name}
-
-			// the execution should be given to the worker
-			ginkgo.By("Waiting to be admitted in worker1 and manager", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sManagerClient.Get(ctx, wlLookupKey, createdLeaderWorkload)).To(gomega.Succeed())
-					g.Expect(workload.FindAdmissionCheck(createdLeaderWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAc.Name))).To(gomega.BeComparableTo(&kueue.AdmissionCheckState{
-						Name:    kueue.AdmissionCheckReference(multiKueueAc.Name),
-						State:   kueue.CheckStateReady,
-						Message: `The workload got reservation on "worker1"`,
-					}, cmpopts.IgnoreFields(kueue.AdmissionCheckState{}, "LastTransitionTime")))
-					g.Expect(apimeta.FindStatusCondition(createdLeaderWorkload.Status.Conditions, kueue.WorkloadAdmitted)).To(gomega.BeComparableTo(&metav1.Condition{
-						Type:    kueue.WorkloadAdmitted,
-						Status:  metav1.ConditionTrue,
-						Reason:  "Admitted",
-						Message: "The workload is admitted",
-					}, util.IgnoreConditionTimestampsAndObservedGeneration))
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
-			})
+			// the execution should be given to the worker1
+			waitForJobAdmitted(wlLookupKey, multiKueueAc.Name, "worker1")
 
 			ginkgo.By("Waiting for the jobSet to get status updates", func() {
 				gomega.Eventually(func(g gomega.Gomega) {
@@ -594,14 +573,8 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the jobSet is completed", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					workerWl := &kueue.Workload{}
-					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					workerJobSet := &jobset.JobSet{}
-					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(jobSet), workerJobSet)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(jobSet), workerJobSet)).To(utiltesting.BeNotFoundError())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				expectObjectToBeDeletedOnWorkerClusters(ctx, createdLeaderWorkload)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, jobSet)
 
 				createdJobSet := &jobset.JobSet{}
 				gomega.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(jobSet), createdJobSet)).To(gomega.Succeed())
@@ -671,14 +644,8 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the appwrapper is completed", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					workerWl := &kueue.Workload{}
-					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					workerAppWrapper := &awv1beta2.AppWrapper{}
-					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(aw), workerAppWrapper)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(aw), workerAppWrapper)).To(utiltesting.BeNotFoundError())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				expectObjectToBeDeletedOnWorkerClusters(ctx, createdWorkload)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, aw)
 
 				createdAppWrapper := &awv1beta2.AppWrapper{}
 				gomega.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(aw), createdAppWrapper)).To(gomega.Succeed())
@@ -739,14 +706,14 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the PyTorchJob is completed", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					workerWl := &kueue.Workload{}
-					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					workerPyTorchJob := &kftraining.PyTorchJob{}
-					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(pyTorchJob), workerPyTorchJob)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(pyTorchJob), workerPyTorchJob)).To(utiltesting.BeNotFoundError())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				wl := &kueue.Workload{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      wlLookupKey.Name,
+						Namespace: wlLookupKey.Namespace,
+					},
+				}
+				expectObjectToBeDeletedOnWorkerClusters(ctx, wl)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, pyTorchJob)
 			})
 		})
 
@@ -801,14 +768,14 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the MPIJob is completed", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					workerWl := &kueue.Workload{}
-					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					workerMPIJob := &kfmpi.MPIJob{}
-					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(mpijob), workerMPIJob)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(mpijob), workerMPIJob)).To(utiltesting.BeNotFoundError())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				wl := &kueue.Workload{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      wlLookupKey.Name,
+						Namespace: wlLookupKey.Namespace,
+					},
+				}
+				expectObjectToBeDeletedOnWorkerClusters(ctx, wl)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, mpijob)
 			})
 		})
 
@@ -845,14 +812,14 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the RayJob is completed", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					workerWl := &kueue.Workload{}
-					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					workerRayJob := &rayv1.RayJob{}
-					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(rayjob), workerRayJob)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(rayjob), workerRayJob)).To(utiltesting.BeNotFoundError())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				wl := &kueue.Workload{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      wlLookupKey.Name,
+						Namespace: wlLookupKey.Namespace,
+					},
+				}
+				expectObjectToBeDeletedOnWorkerClusters(ctx, wl)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, rayjob)
 			})
 		})
 
@@ -968,14 +935,8 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the job is completed", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					workerWl := &kueue.Workload{}
-					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
-					workerJob := &batchv1.Job{}
-					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(job), workerJob)).To(utiltesting.BeNotFoundError())
-					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(job), workerJob)).To(utiltesting.BeNotFoundError())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				expectObjectToBeDeletedOnWorkerClusters(ctx, createdLeaderWorkload)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, job)
 
 				createdJob := &batchv1.Job{}
 				gomega.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
@@ -1126,3 +1087,15 @@ func GetMultiKueueClusterNameFromAdmissionCheckMessage(message string) string {
 	}
 	return ""
 }
+
+type objAsPtr[T any] interface {
+	client.Object
+	*T
+}
+
+func expectObjectToBeDeletedOnWorkerClusters[PtrT objAsPtr[T], T any](ctx context.Context, obj PtrT) {
+	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
+		util.ExpectObjectToBeDeleted(ctx, k8sWorker1Client, obj, false)
+		util.ExpectObjectToBeDeleted(ctx, k8sWorker2Client, obj, false)
+	}, util.Timeout, util.Interval).Should(gomega.Succeed())
+}
diff --git a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
index da992d79e..dce85f0ae 100644
--- a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
+++ b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
@@ -1300,6 +1300,189 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 		})
 	})
 
+	ginkgo.When("Workload uses a provision admission check with BackoffLimitCount=2", func() {
+		var (
+			ns             *corev1.Namespace
+			wlKey          types.NamespacedName
+			ac             *kueue.AdmissionCheck
+			prc            *kueue.ProvisioningRequestConfig
+			rf             *kueue.ResourceFlavor
+			cq             *kueue.ClusterQueue
+			lq             *kueue.LocalQueue
+			admission      *kueue.Admission
+			createdRequest autoscaling.ProvisioningRequest
+			updatedWl      kueue.Workload
+			provReqKey     types.NamespacedName
+		)
+		ginkgo.JustBeforeEach(func() {
+			ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "provisioning-")
+
+			prc = baseConfig.Clone().RetryLimit(2).BaseBackoff(2).Obj()
+			util.MustCreate(ctx, k8sClient, prc)
+
+			ac = testing.MakeAdmissionCheck("ac-prov").
+				ControllerName(kueue.ProvisioningRequestControllerName).
+				Parameters(kueue.GroupVersion.Group, "ProvisioningRequestConfig", prc.Name).
+				Obj()
+			util.MustCreate(ctx, k8sClient, ac)
+
+			rf = testing.MakeResourceFlavor("rf1").Obj()
+			util.MustCreate(ctx, k8sClient, rf)
+
+			cq = testing.MakeClusterQueue("cluster-queue").
+				ResourceGroup(*testing.MakeFlavorQuotas(rf.Name).
+					Resource(resourceGPU, "5").Obj()).
+				AdmissionChecks(kueue.AdmissionCheckReference(ac.Name)).
+				Obj()
+			util.MustCreate(ctx, k8sClient, cq)
+
+			lq = testing.MakeLocalQueue("queue", ns.Name).ClusterQueue(cq.Name).Obj()
+			util.MustCreate(ctx, k8sClient, lq)
+
+			wl := testing.MakeWorkload("wl", ns.Name).
+				Queue(kueue.LocalQueueName(lq.Name)).
+				PodSets(
+					*testing.MakePodSet("ps1", 3).
+						Request(corev1.ResourceCPU, "1").
+						Image("image").
+						Obj(),
+				).Obj()
+			util.MustCreate(ctx, k8sClient, wl)
+
+			wlKey = client.ObjectKeyFromObject(wl)
+			admission = testing.MakeAdmission(cq.Name).
+				PodSets(
+					kueue.PodSetAssignment{
+						Name: "ps1",
+						Flavors: map[corev1.ResourceName]kueue.ResourceFlavorReference{
+							corev1.ResourceCPU: kueue.ResourceFlavorReference(rf.Name),
+						},
+						ResourceUsage: map[corev1.ResourceName]resource.Quantity{
+							corev1.ResourceCPU: resource.MustParse("3"),
+						},
+						Count: ptr.To[int32](3),
+					}).Obj()
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, cq, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, rf, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, ac, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, prc, true)
+		})
+
+		ginkgo.It("Should retry twice if a ProvisioningRequest fails twice", func() {
+			ginkgo.By("Setting the quota reservation to the workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).Should(gomega.Succeed())
+					g.Expect(util.SetQuotaReservation(ctx, k8sClient, &updatedWl, admission)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Setting the provision request-1 as Failed", func() {
+				provReqKey = types.NamespacedName{
+					Namespace: wlKey.Namespace,
+					Name:      provisioning.ProvisioningRequestName(wlKey.Name, kueue.AdmissionCheckReference(ac.Name), 1),
+				}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, provReqKey, &createdRequest)).Should(gomega.Succeed())
+					apimeta.SetStatusCondition(&createdRequest.Status.Conditions, metav1.Condition{
+						Type:   autoscaling.Failed,
+						Status: metav1.ConditionTrue,
+						Reason: autoscaling.Failed,
+					})
+					g.Expect(k8sClient.Status().Update(ctx, &createdRequest)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking the AdmissionCheck is set to Retry, and the workload has requeueState set", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
+					g.Expect(updatedWl.Status.RequeueState).NotTo(gomega.BeNil())
+					g.Expect(*updatedWl.Status.RequeueState.Count).To(gomega.Equal(int32(1)))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking the Workload is Evicted, and all AdmissionChecks are reset to Pending", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
+					_, evicted := workload.IsEvictedByAdmissionCheck(&updatedWl)
+					g.Expect(evicted).To(gomega.BeTrue())
+					check := workload.FindAdmissionCheck(updatedWl.Status.AdmissionChecks, kueue.AdmissionCheckReference(ac.Name))
+					g.Expect(check).NotTo(gomega.BeNil())
+					g.Expect(check.State).To(gomega.Equal(kueue.CheckStatePending))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Setting the quota reservation to the workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).Should(gomega.Succeed())
+					g.Expect(util.SetQuotaReservation(ctx, k8sClient, &updatedWl, admission)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking the provision request-2 exists", func() {
+				gomega.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
+				provReqKey = types.NamespacedName{
+					Namespace: wlKey.Namespace,
+					Name:      provisioning.ProvisioningRequestName(wlKey.Name, kueue.AdmissionCheckReference(ac.Name), 2),
+				}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, provReqKey, &createdRequest)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Setting the provision request-2 as Failed", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, provReqKey, &createdRequest)).Should(gomega.Succeed())
+					apimeta.SetStatusCondition(&createdRequest.Status.Conditions, metav1.Condition{
+						Type:   autoscaling.Failed,
+						Status: metav1.ConditionTrue,
+						Reason: autoscaling.Failed,
+					})
+					g.Expect(k8sClient.Status().Update(ctx, &createdRequest)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking the AdmissionCheck is set to Retry, and the workload has requeueState set", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
+					g.Expect(updatedWl.Status.RequeueState).NotTo(gomega.BeNil())
+					g.Expect(*updatedWl.Status.RequeueState.Count).To(gomega.Equal(int32(2)))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking the Workload is Evicted, and all AdmissionChecks are reset to Pending", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
+					_, evicted := workload.IsEvictedByAdmissionCheck(&updatedWl)
+					g.Expect(evicted).To(gomega.BeTrue())
+					check := workload.FindAdmissionCheck(updatedWl.Status.AdmissionChecks, kueue.AdmissionCheckReference(ac.Name))
+					g.Expect(check).NotTo(gomega.BeNil())
+					g.Expect(check.State).To(gomega.Equal(kueue.CheckStatePending))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Setting the quota reservation to the workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).Should(gomega.Succeed())
+					g.Expect(util.SetQuotaReservation(ctx, k8sClient, &updatedWl, admission)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking the provision request-3 exists", func() {
+				provReqKey = types.NamespacedName{
+					Namespace: wlKey.Namespace,
+					Name:      provisioning.ProvisioningRequestName(wlKey.Name, kueue.AdmissionCheckReference(ac.Name), 3),
+				}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, provReqKey, &createdRequest)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+
 	ginkgo.When("A workload is using a provision request with enabled PodSetMergePolicy", func() {
 		var (
 			ns             *corev1.Namespace
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index 29b799692..5fe44f97e 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -1026,6 +1026,123 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					}, util.Timeout, util.Interval).Should(gomega.Succeed())
 				})
 			})
+			ginkgo.It("should clear the annotation when node recovers", func() {
+				var wl1 *kueue.Workload
+				nodeName := nodes[0].Name
+
+				ginkgo.By("creating a workload", func() {
+					wl1 = testing.MakeWorkload("wl1", ns.Name).
+						PodSets(*testing.MakePodSet("worker", 2).
+							// requiring the same block makes sure that no replacement is possible
+							RequiredTopologyRequest(testing.DefaultBlockTopologyLevel).
+							Obj()).
+						Queue(kueue.LocalQueueName(localQueue.Name)).Request(corev1.ResourceCPU, "1").Obj()
+					util.MustCreate(ctx, k8sClient, wl1)
+				})
+
+				ginkgo.By("verify the workload is admitted", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+					gomega.Expect(wl1.Status.Admission.PodSetAssignments[0].TopologyAssignment).Should(gomega.BeComparableTo(
+						&kueue.TopologyAssignment{
+							Levels: []string{corev1.LabelHostname},
+							Domains: []kueue.TopologyDomainAssignment{
+								{Count: 1, Values: []string{"x1"}},
+								{Count: 1, Values: []string{"x2"}},
+							},
+						},
+					))
+				})
+
+				ginkgo.By("making the node NotReady 30s in the past", func() {
+					nodeToUpdate := &corev1.Node{}
+					gomega.Expect(k8sClient.Get(ctx, apitypes.NamespacedName{Name: nodeName}, nodeToUpdate)).Should(gomega.Succeed())
+
+					util.SetNodeCondition(ctx, k8sClient, nodeToUpdate, &corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionFalse,
+						LastTransitionTime: metav1.NewTime(time.Now().Add(-tas.NodeFailureDelay)),
+					})
+				})
+
+				ginkgo.By("verify the workload eventually gets the NodeToReplaceAnnotation", func() {
+					gomega.Eventually(func(g gomega.Gomega) map[string]string {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+						return wl1.Annotations
+					}, util.Timeout, util.Interval).Should(gomega.HaveKeyWithValue(kueuealpha.NodeToReplaceAnnotation, nodeName))
+				})
+
+				ginkgo.By("node recovers", func() {
+					nodeToUpdate := &corev1.Node{}
+					gomega.Expect(k8sClient.Get(ctx, apitypes.NamespacedName{Name: nodeName}, nodeToUpdate)).Should(gomega.Succeed())
+
+					util.SetNodeCondition(ctx, k8sClient, nodeToUpdate, &corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastTransitionTime: metav1.NewTime(time.Now()),
+					})
+				})
+
+				ginkgo.By("verify the annotation is cleared", func() {
+					gomega.Eventually(func(g gomega.Gomega) map[string]string {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+						return wl1.Annotations
+					}, util.Timeout, util.Interval).ShouldNot(gomega.HaveKeyWithValue(kueuealpha.NodeToReplaceAnnotation, nodeName))
+				})
+			})
+
+			ginkgo.It("should clear the annotation when node reappears", func() {
+				var wl1 *kueue.Workload
+				nodeName := nodes[0].Name
+
+				ginkgo.By("creating a workload", func() {
+					wl1 = testing.MakeWorkload("wl1", ns.Name).
+						PodSets(*testing.MakePodSet("worker", 2).
+							RequiredTopologyRequest(testing.DefaultBlockTopologyLevel).
+							Obj()).
+						Queue(kueue.LocalQueueName(localQueue.Name)).Request(corev1.ResourceCPU, "1").Obj()
+					util.MustCreate(ctx, k8sClient, wl1)
+				})
+
+				ginkgo.By("verify the workload is admitted", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+					gomega.Expect(wl1.Status.Admission.PodSetAssignments[0].TopologyAssignment).Should(gomega.BeComparableTo(
+						&kueue.TopologyAssignment{
+							Levels: []string{corev1.LabelHostname},
+							Domains: []kueue.TopologyDomainAssignment{
+								{Count: 1, Values: []string{"x1"}},
+								{Count: 1, Values: []string{"x2"}},
+							},
+						},
+					))
+				})
+
+				nodeToDelete := &corev1.Node{}
+				ginkgo.By("deleting the node", func() {
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodeName}, nodeToDelete)).Should(gomega.Succeed())
+					gomega.Expect(k8sClient.Delete(ctx, nodeToDelete)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the workload eventually gets the NodeToReplaceAnnotation", func() {
+					gomega.Eventually(func(g gomega.Gomega) map[string]string {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+						return wl1.Annotations
+					}, util.Timeout, util.Interval).Should(gomega.HaveKeyWithValue(kueuealpha.NodeToReplaceAnnotation, nodeName))
+				})
+
+				ginkgo.By("node reappears", func() {
+					nodeToDelete.ResourceVersion = ""
+					gomega.Expect(k8sClient.Create(ctx, nodeToDelete)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the annotation is cleared", func() {
+					gomega.Eventually(func(g gomega.Gomega) map[string]string {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+						return wl1.Annotations
+					}, util.Timeout, util.Interval).ShouldNot(gomega.HaveKeyWithValue(kueuealpha.NodeToReplaceAnnotation, nodeName))
+				})
+			})
 
 			ginkgo.It("should update workload TopologyAssignment after a node recovers", func() {
 				var wl1, wl2 *kueue.Workload