Repository: kubernetes-sigs/kueue

Issue:
[TAS] Node failure controller should delete nodeToReplace annotation if a node recovers/reappears
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:
Let's assume a Workload uses the whole capacity in a cluster. After node failure, a Workload gets annotated with nodeToReplace annotation. Let's say a node has then recovered and Workload can run as usual. After that Workload still have the annotation even if, there's no need to replace the node. It has a few consequences. First of all we waste scheduling cycle for a Workload that doesn't need the replacement. Secondly, if another node fails then the Workload will gets evicted while it shouldn't.

I'd like add a symmetric mechanism to current node failure controller, so it deletes the annotation if a node recovers/reappears.

**Why is this needed**:

**Completion requirements**:

This enhancement requires the following artifacts:

- [ ] Design doc
- [ ] API change
- [ ] Docs update

The artifacts should be linked in subsequent comments.


Discussion:
@PBundyra: /kind bug
/remove label feature@PBundyra: /remove-kind feature@PBundyra: /cc @pajakd @mimowo @mwysokin @mimowo: +1@pajakd: Yes, definitely we have to fix this. One thing to keep in mind will be the time complexity -- in each reconcile of the node failure controller, we list all the workloads and if we will want to reconcile on each update of healthy nodes this might take some time. But overall I agree.@PBundyra: > Yes, definitely we have to fix this. One thing to keep in mind will be the time complexity -- in each reconcile of the node failure controller, we list all the workloads and if we will want to reconcile on each update of healthy nodes this might take some time. But overall I agree.

I'd say we should list only on an Update that changes status of a node from Ready=False -> Ready=True or on a Create event. In this case I think the time complexity is acceptable

@pajakd: /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/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