Repository: kubernetes-sigs/kueue

Issue:
[Flaky] Scheduler when Using AdmissionFairSharing at Cohort level should preempt a workload from LQ with higher recent usage
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

If the matter is security related, please disclose it privately via https://kubernetes.io/security/
-->


**What happened**:
Scheduler Fair Sharing Suite: [It] Scheduler when Using AdmissionFairSharing at Cohort level should preempt a workload from LQ with higher recent usage

```
{Timed out after 10.001s.
The function passed to Eventually failed at /home/prow/go/src/sigs.k8s.io/kueue/test/util/util.go:431 with:
Not enough workloads are preempted
Expected
    <int>: 0
to equal
    <int>: 1 failed [FAILED] Timed out after 10.001s.
The function passed to Eventually failed at /home/prow/go/src/sigs.k8s.io/kueue/test/util/util.go:431 with:
Not enough workloads are preempted
Expected
    <int>: 0
to equal
    <int>: 1
In [It] at: /home/prow/go/src/sigs.k8s.io/kueue/test/integration/singlecluster/scheduler/fairsharing/fair_sharing_test.go:508 @ 07/28/25 09:04:04.895
}
```

**What you expected to happen**:
No errors

**How to reproduce it (as minimally and precisely as possible)**:
https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/5174/pull-kueue-test-integration-baseline-main/1949754497840451584

**Anything else we need to know?**:

/kind flake

**Environment**:
- Kubernetes version (use `kubectl version`):
- Kueue version (use `git describe --tags --dirty --always`):
- Cloud provider or hardware configuration:
- OS (e.g: `cat /etc/os-release`):
- Kernel (e.g. `uname -a`):
- Install tools:
- Others:

Discussion:
@mimowo: cc @PBundyra ptal@mimowo: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6283/pull-kueue-test-integration-baseline-main/1950559248463171584@IrvingMg: /assign@mimowo: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6297/pull-kueue-test-integration-baseline-main/1950867693955452928 @IrvingMg: > https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6297/pull-kueue-test-integration-baseline-main/1950867693955452928

Actually, this is a different test. I've opened #6324.
@mimowo: https://prow.k8s.io/view/gs/kubernetes-ci-logs/logs/periodic-kueue-test-integration-main/1951171153938616320
happened on nightly@mbobrovskyi: One more https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6345/pull-kueue-test-integration-baseline-main/1951184148903235584

Solution patch:
diff --git a/config/components/certmanager/certificate-metrics.yaml b/config/components/certmanager/certificate-metrics.yaml
index afdab84cc..198748382 100644
--- a/config/components/certmanager/certificate-metrics.yaml
+++ b/config/components/certmanager/certificate-metrics.yaml
@@ -3,7 +3,7 @@
 apiVersion: cert-manager.io/v1
 kind: Certificate
 metadata:
-  name: metrics-certs  # this name should match the one appeared in kustomizeconfig.yaml
+  name: metrics-cert  # this name should match the one appeared in kustomizeconfig.yaml
   namespace: system
 spec:
   commonName: metrics
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/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/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 639cbc443..d7ab5f4bd 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -255,7 +255,7 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		// exception due to complicated logic, not using PrepareForEviction (temporarily)
 		workload.ResetClusterNomination(&wl)
 		updated = workload.ResetChecksOnEviction(&wl, r.clock.Now()) || updated
-		reportWorkloadEvictedOnce := workload.WorkloadEvictionStateInc(&wl, kueue.WorkloadDeactivated, "")
+		reportWorkloadEvictedOnce := workload.WorkloadEvictionStateInc(&wl, reason, "")
 		if updated {
 			if err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock); err != nil {
 				if apierrors.IsNotFound(err) {
@@ -266,7 +266,7 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			if evicted && wl.Status.Admission != nil {
 				workload.ReportEvictedWorkload(r.recorder, &wl, wl.Status.Admission.ClusterQueue, reason, message)
 				if reportWorkloadEvictedOnce {
-					metrics.ReportEvictedWorkloadsOnce(wl.Status.Admission.ClusterQueue, kueue.WorkloadDeactivated, "")
+					metrics.ReportEvictedWorkloadsOnce(wl.Status.Admission.ClusterQueue, reason, "")
 				}
 			}
 			return ctrl.Result{}, nil
@@ -488,12 +488,16 @@ func (r *WorkloadReconciler) reconcileOnLocalQueueActiveState(ctx context.Contex
 		}
 		log.V(3).Info("Workload is evicted because the LocalQueue is stopped", "localQueue", klog.KRef(wl.Namespace, string(wl.Spec.QueueName)))
 		workload.PrepareForEviction(wl, r.clock.Now(), kueue.WorkloadEvictedByLocalQueueStopped, "The LocalQueue is stopped")
+		reportWorkloadEvictedOnce := workload.WorkloadEvictionStateInc(wl, kueue.WorkloadEvictedByLocalQueueStopped, "")
 		err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock)
 		if err == nil {
 			cqName := lq.Spec.ClusterQueue
 			if slices.Contains(r.queues.GetClusterQueueNames(), cqName) {
 				workload.ReportEvictedWorkload(r.recorder, wl, cqName, kueue.WorkloadEvictedByLocalQueueStopped, "The LocalQueue is stopped")
 			}
+			if reportWorkloadEvictedOnce {
+				metrics.ReportEvictedWorkloadsOnce(cqName, kueue.WorkloadEvictedByLocalQueueStopped, "")
+			}
 		}
 		return true, err
 	}
diff --git a/pkg/controller/tas/node_failure_controller.go b/pkg/controller/tas/node_failure_controller.go
index 5ec8513a4..fbbcfb61e 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
 }
 
@@ -302,12 +300,7 @@ func (r *nodeFailureReconciler) patchWorkloadsForNodeToReplace(ctx context.Conte
 }
 
 func (r *nodeFailureReconciler) startEviction(ctx context.Context, wl *kueue.Workload, evictionMessage string) error {
-	workload.PrepareForEviction(wl, r.clock.Now(), kueue.WorkloadEvictedDueToNodeFailures, evictionMessage)
-	if err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock); err != nil {
-		return err
-	}
-	workload.ReportEvictedWorkload(r.recorder, wl, wl.Status.Admission.ClusterQueue, kueue.WorkloadEvictedDueToNodeFailures, evictionMessage)
-	return nil
+	return workload.EvictWorkload(ctx, r.client, r.recorder, wl, kueue.WorkloadEvictedDueToNodeFailures, evictionMessage, r.clock)
 }
 
 func (r *nodeFailureReconciler) reconcileForReplaceNodeOnPodTermination(ctx context.Context, nodeName string) (ctrl.Result, error) {
@@ -323,3 +316,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/metrics/metrics.go b/pkg/metrics/metrics.go
index 948de6a87..9e16e9087 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -164,7 +164,12 @@ The label 'reason' can have the following values:
 - "PodsReadyTimeout" means that the eviction took place due to a PodsReady timeout.
 - "AdmissionCheck" means that the workload was evicted because at least one admission check transitioned to False.
 - "ClusterQueueStopped" means that the workload was evicted because the ClusterQueue is stopped.
-- "Deactivated" means that the workload was evicted because spec.active is set to false`,
+- "LocalQueueStopped" means that the workload was evicted because the LocalQueue is stopped.
+- "NodeFailures" means that the workload was evicted due to node failures when using TopologyAwareScheduling.
+- "Deactivated" means that the workload was evicted because spec.active is set to false.
+- "DeactivatedDueToAdmissionCheck" means that the workload was evicted and deactivated by Kueue due to a rejected admission check.
+- "DeactivatedDueToMaximumExecutionTimeExceeded" means that the workload was evicted and deactivated by Kueue due to maximum execution time exceeded.
+- "DeactivatedDueToRequeuingLimitExceeded" means that the workload was evicted and deactivated by Kueue due to requeuing limit exceeded.`,
 			Buckets: generateExponentialBuckets(14),
 		}, []string{"cluster_queue", "reason"},
 	)
@@ -276,7 +281,12 @@ The label 'reason' can have the following values:
 - "PodsReadyTimeout" means that the eviction took place due to a PodsReady timeout.
 - "AdmissionCheck" means that the workload was evicted because at least one admission check transitioned to False.
 - "ClusterQueueStopped" means that the workload was evicted because the ClusterQueue is stopped.
-- "Deactivated" means that the workload was evicted because spec.active is set to false`,
+- "LocalQueueStopped" means that the workload was evicted because the LocalQueue is stopped.
+- "NodeFailures" means that the workload was evicted due to node failures when using TopologyAwareScheduling.
+- "Deactivated" means that the workload was evicted because spec.active is set to false.
+- "DeactivatedDueToAdmissionCheck" means that the workload was evicted and deactivated by Kueue due to a rejected admission check.
+- "DeactivatedDueToMaximumExecutionTimeExceeded" means that the workload was evicted and deactivated by Kueue due to maximum execution time exceeded.
+- "DeactivatedDueToRequeuingLimitExceeded" means that the workload was evicted and deactivated by Kueue due to requeuing limit exceeded.`,
 		}, []string{"cluster_queue", "reason"},
 	)
 
@@ -298,7 +308,12 @@ The label 'reason' can have the following values:
 - "PodsReadyTimeout" means that the eviction took place due to a PodsReady timeout.
 - "AdmissionCheck" means that the workload was evicted because at least one admission check transitioned to False.
 - "ClusterQueueStopped" means that the workload was evicted because the ClusterQueue is stopped.
-- "Deactivated" means that the workload was evicted because spec.active is set to false`,
+- "LocalQueueStopped" means that the workload was evicted because the LocalQueue is stopped.
+- "NodeFailures" means that the workload was evicted due to node failures when using TopologyAwareScheduling.
+- "Deactivated" means that the workload was evicted because spec.active is set to false.
+- "DeactivatedDueToAdmissionCheck" means that the workload was evicted and deactivated by Kueue due to a rejected admission check.
+- "DeactivatedDueToMaximumExecutionTimeExceeded" means that the workload was evicted and deactivated by Kueue due to maximum execution time exceeded.
+- "DeactivatedDueToRequeuingLimitExceeded" means that the workload was evicted and deactivated by Kueue due to requeuing limit exceeded.`,
 		}, []string{"name", "namespace", "reason"},
 	)
 
@@ -312,7 +327,12 @@ The label 'reason' can have the following values:
 - "PodsReadyTimeout" means that the eviction took place due to a PodsReady timeout.
 - "AdmissionCheck" means that the workload was evicted because at least one admission check transitioned to False.
 - "ClusterQueueStopped" means that the workload was evicted because the ClusterQueue is stopped.
-- "Deactivated" means that the workload was evicted because spec.active is set to false`,
+- "LocalQueueStopped" means that the workload was evicted because the LocalQueue is stopped.
+- "NodeFailures" means that the workload was evicted due to node failures when using TopologyAwareScheduling.
+- "Deactivated" means that the workload was evicted because spec.active is set to false.
+- "DeactivatedDueToAdmissionCheck" means that the workload was evicted and deactivated by Kueue due to a rejected admission check.
+- "DeactivatedDueToMaximumExecutionTimeExceeded" means that the workload was evicted and deactivated by Kueue due to maximum execution time exceeded.
+- "DeactivatedDueToRequeuingLimitExceeded" means that the workload was evicted and deactivated by Kueue due to requeuing limit exceeded.`,
 		}, []string{"cluster_queue", "reason", "detailed_reason"},
 	)
 
@@ -443,8 +463,8 @@ For a LocalQueue, the metric only reports a value of 1 for one of the statuses.`
 		prometheus.GaugeOpts{
 			Subsystem: constants.KueueName,
 			Name:      "cluster_queue_weighted_share",
-			Help: `Reports a value that representing the maximum of the ratios of usage above nominal 
-quota to the lendable resources in the cohort, among all the resources provided by 
+			Help: `Reports a value that representing the maximum of the ratios of usage above nominal
+quota to the lendable resources in the cohort, among all the resources provided by
 the ClusterQueue, and divided by the weight.
 If zero, it means that the usage of the ClusterQueue is below the nominal quota.
 If the ClusterQueue has a weight of zero and is borrowing, this will return 9223372036854775807,
@@ -456,8 +476,8 @@ the maximum possible share value.`,
 		prometheus.GaugeOpts{
 			Subsystem: constants.KueueName,
 			Name:      "cohort_weighted_share",
-			Help: `Reports a value that representing the maximum of the ratios of usage above nominal 
-quota to the lendable resources in the Cohort, among all the resources provided by 
+			Help: `Reports a value that representing the maximum of the ratios of usage above nominal
+quota to the lendable resources in the Cohort, among all the resources provided by
 the Cohort, and divided by the weight.
 If zero, it means that the usage of the Cohort is below the nominal quota.
 If the Cohort has a weight of zero and is borrowing, this will return 9223372036854775807,
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..ef92ed7c3 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -188,10 +188,13 @@ func (p *Preemptor) applyPreemptionWithSSA(ctx context.Context, w *kueue.Workloa
 	workload.PrepareForEviction(w, p.clock.Now(), kueue.WorkloadEvictedByPreemption, message)
 	reportWorkloadEvictedOnce := workload.WorkloadEvictionStateInc(w, kueue.WorkloadEvictedByPreemption, "")
 	workload.SetPreemptedCondition(w, reason, message)
-	if reportWorkloadEvictedOnce {
-		metrics.ReportEvictedWorkloadsOnce(w.Status.Admission.ClusterQueue, kueue.WorkloadEvictedByPreemption, "")
+	err := workload.ApplyAdmissionStatus(ctx, p.client, w, true, p.clock)
+	if err == nil {
+		if reportWorkloadEvictedOnce {
+			metrics.ReportEvictedWorkloadsOnce(w.Status.Admission.ClusterQueue, kueue.WorkloadEvictedByPreemption, "")
+		}
 	}
-	return workload.ApplyAdmissionStatus(ctx, p.client, w, true, p.clock)
+	return err
 }
 
 type preemptionAttemptOpts struct {
@@ -208,6 +211,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 +384,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 +546,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 +560,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/zh-CN/docs/tasks/run/multikueue/deployment.md b/site/content/zh-CN/docs/tasks/run/multikueue/deployment.md
index c3a166359..6c699cf07 100644
--- a/site/content/zh-CN/docs/tasks/run/multikueue/deployment.md
+++ b/site/content/zh-CN/docs/tasks/run/multikueue/deployment.md
@@ -1,29 +1,27 @@
 ---
-title: "Run Deployment in Multi-Cluster"
+title: "在多集群环境中运行 Deployment"
 linkTitle: "Deployment"
 weight: 2
 date: 2025-02-17
-description: >
-  Run a MultiKueue scheduled Deployment.
+description: 运行 MultiKueue 调度的 Deployment。
 ---
 
-## Before you begin
+## 开始之前 {#before-you-begin}
 
-1. Check the [MultiKueue installation guide](/docs/tasks/manage/setup_multikueue) on how to properly setup MultiKueue clusters.
+1. 请查阅 [MultiKueue 安装指南](/zh-CN/docs/tasks/manage/setup_multikueue)了解如何正确设置 MultiKueue 集群。
 
-2. Follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin)
-to learn how to enable and configure the `pod` integration which is required for enabling the `deployment` integration.
+2. 按照[运行普通 Pod](/zh-CN/docs/tasks/run/plain_pods/#before-you-begin)的步骤，了解如何启用和配置 Pod 集成，这对于启用 Deployment 集成是必需的。
 
-Deployments receive live status updates through the status from the remote Pods created on the worker cluster.
+Deployment 会实时接收在工作集群创建的远程 Pod 的状态，并更新。
 
 {{< feature-state state="beta" for_version="v0.11.0" >}}
 
-{{% alert title="Note" color="primary" %}}
-In this current implementation, when Deployments are created in environments with multiple worker clusters, Pods are allocated to any worker.
+{{% alert title="注意" color="primary" %}}
+在当前实现中，当在有多个工作集群的环境中创建 Deployment 时，Pod 会被分配到任意工作集群。
 {{% /alert %}}
 
-## Example
+## 示例 {#examples}
 
-Once the setup is complete you can test it by running the example below:
+完成设置后，你可以通过运行以下示例进行测试：
 
 {{< include "examples/serving-workloads/sample-deployment.yaml" "yaml" >}}
diff --git a/site/content/zh-CN/docs/tasks/run/multikueue/job.md b/site/content/zh-CN/docs/tasks/run/multikueue/job.md
index a1a7e7f8e..ac40d8d67 100644
--- a/site/content/zh-CN/docs/tasks/run/multikueue/job.md
+++ b/site/content/zh-CN/docs/tasks/run/multikueue/job.md
@@ -1,47 +1,46 @@
 ---
-title: "Run Kubernetes Job in Multi-Cluster"
+title: "在多集群环境中运行 Job"
 linkTitle: "Kubernetes Job"
 weight: 2
 date: 2024-11-05
-description: >
-  Run a MultiKueue scheduled Kubernetes Job.
+description: 运行 MultiKueue 调度的 Kubernetes 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.8.1.
+为方便安装和使用，建议使用 Kueue v0.8.1 以上版本。
 
-The recommended way of running MultiKueue depends on the configuration of the `JobManagedBy` feature gate in your cluster. 
+运行 MultiKueue 的推荐方式取决于集群中 `JobManagedBy` 特性门控的配置。
 
-{{% alert title="Note" color="primary" %}}
-The `JobManagedBy` feature gate is disabled in 1.30 and 1.31 by default, and will be enabled in 1.32 by default.
+{{% alert title="注意" color="primary" %}}
+`JobManagedBy` 特性门控在 Kubernetes v1.30 和 v1.31 版本中默认禁用，在 v1.32 版本中默认启用。
 {{% /alert %}}
 
-### Cluster with JobManagedBy enabled
+### 集群启用 JobManagedBy {#cluster-with-jobmanagedby-enabled}
 
-When `JobManagedBy` is enabled in your cluster we recommend configuring Kueue to enable the `MultiKueueBatchJobWithManagedBy` feature gate. 
+当集群启用 `JobManagedBy` 时，建议配置 Kueue 启用 `MultiKueueBatchJobWithManagedBy` 特性门控。
 
-When `MultiKueueBatchJobWithManagedBy` is enabled, the current status of a Job being executed by MultiKueue on a worker cluster is live-updated on the management cluster.
+启用后，MultiKueue 在工作集群上执行的 Job 的当前状态会实时同步到管理集群。
 
-This gives the users and automation tools the ability to track the progress of Job status (.status) without lookup to the
-worker cluster, making MultiKueue transparent from that perspective.
+这样，用户和自动化工具可以在不访问工作集群的情况下，直接跟踪 Job 的状态（`.status`），
+从而实现 MultiKueue 的透明化。
 
-### Cluster with JobManagedBy disabled
+### 集群未启用 JobManagedBy {#cluster-with-jobmanagedby-disabled}
 
-When `JobManagedBy` is disabled in your cluster you should make sure `MultiKueueBatchJobWithManagedBy` is also disabled in Kueue. 
+当集群未启用 `JobManagedBy` 时，应确保 Kueue 也未启用 `MultiKueueBatchJobWithManagedBy`。
 
-This is important so that MultiKueue does not conflict with the build-in Job controller on the management cluster. 
+这样可以避免 MultiKueue 与管理集群上的内置 Job 控制器发生冲突。
 
-As a limitation of this deployment mode, to get the actual status of the job, you need to access the worker cluster.
+这种部署模式有一个缺点，你需要访问工作集群才能获取 Job 的实际状态。
 
-You can identify the worker cluster running the job by checking the AC status message of the workload object in the management cluster.
+你可以通过检查管理集群中工作负载对象的 AC 状态消息，辨别正在运行 Job 的工作集群。
 
-Also, the job is suspended from the perspective of the management cluster until it is `Finished`.
+此外，从管理集群的角度看，Job 在达到 `Finished` 之前，会一直处于挂起状态。
 
-## Example
+## 示例 {#examples}
 
-Once the setup is complete you can test it by running the example below:
+完成设置后，你可以通过运行以下示例进行测试：
 
 {{< include "examples/jobs/sample-job.yaml" "yaml" >}}



Test patch:
diff --git a/pkg/controller/core/workload_controller_test.go b/pkg/controller/core/workload_controller_test.go
index 13130afa6..b69709c23 100644
--- a/pkg/controller/core/workload_controller_test.go
+++ b/pkg/controller/core/workload_controller_test.go
@@ -677,7 +677,7 @@ func TestReconcile(t *testing.T) {
 				).
 				SchedulingStatsEviction(
 					kueue.WorkloadSchedulingStatsEviction{
-						Reason: kueue.WorkloadDeactivated,
+						Reason: "DeactivatedDueToAdmissionCheck",
 						Count:  1,
 					},
 				).
@@ -1344,7 +1344,7 @@ func TestReconcile(t *testing.T) {
 					Message: "exceeding the maximum number of re-queuing retries",
 				}).
 				SchedulingStatsEviction(
-					kueue.WorkloadSchedulingStatsEviction{Reason: kueue.WorkloadDeactivated, Count: 1},
+					kueue.WorkloadSchedulingStatsEviction{Reason: "DeactivatedDueToRequeuingLimitExceeded", Count: 1},
 				).
 				Obj(),
 			wantEvents: []utiltesting.EventRecord{
@@ -1408,7 +1408,7 @@ func TestReconcile(t *testing.T) {
 				}).
 				SchedulingStatsEviction(
 					kueue.WorkloadSchedulingStatsEviction{
-						Reason: kueue.WorkloadDeactivated,
+						Reason: "DeactivatedDueToRequeuingLimitExceeded",
 						Count:  1,
 					},
 				).
@@ -1477,7 +1477,7 @@ func TestReconcile(t *testing.T) {
 				RequeueState(ptr.To[int32](100), nil).
 				SchedulingStatsEviction(
 					kueue.WorkloadSchedulingStatsEviction{
-						Reason: kueue.WorkloadDeactivated,
+						Reason: "DeactivatedDueToRequeuingLimitExceeded",
 						Count:  1,
 					},
 				).
@@ -1546,7 +1546,7 @@ func TestReconcile(t *testing.T) {
 				RequeueState(ptr.To[int32](100), nil).
 				SchedulingStatsEviction(
 					kueue.WorkloadSchedulingStatsEviction{
-						Reason: kueue.WorkloadDeactivated,
+						Reason: "DeactivatedDueToRequeuingLimitExceeded",
 						Count:  1,
 					},
 				).
@@ -1640,6 +1640,12 @@ func TestReconcile(t *testing.T) {
 					Reason:  kueue.WorkloadEvictedByLocalQueueStopped,
 					Message: "The LocalQueue is stopped",
 				}).
+				SchedulingStatsEviction(
+					kueue.WorkloadSchedulingStatsEviction{
+						Reason: kueue.WorkloadEvictedByLocalQueueStopped,
+						Count:  1,
+					},
+				).
 				Obj(),
 			wantEvents: []utiltesting.EventRecord{
 				{
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/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index 371e36edf..f85e849c4 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -177,12 +177,12 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 		util.ExpectObjectToBeDeleted(ctx, k8sWorker2Client, worker2Cq, true)
 		util.ExpectObjectToBeDeleted(ctx, k8sWorker2Client, worker2Flavor, true)
 
-		util.ExpectObjectToBeDeleted(ctx, k8sManagerClient, managerCq, true)
-		util.ExpectObjectToBeDeleted(ctx, k8sManagerClient, managerFlavor, true)
-		util.ExpectObjectToBeDeleted(ctx, k8sManagerClient, multiKueueAc, true)
-		util.ExpectObjectToBeDeleted(ctx, k8sManagerClient, multiKueueConfig, true)
-		util.ExpectObjectToBeDeleted(ctx, k8sManagerClient, workerCluster1, true)
-		util.ExpectObjectToBeDeleted(ctx, k8sManagerClient, workerCluster2, true)
+		util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sManagerClient, managerCq, true, util.LongTimeout)
+		util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sManagerClient, managerFlavor, true, util.LongTimeout)
+		util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sManagerClient, multiKueueAc, true, util.LongTimeout)
+		util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sManagerClient, multiKueueConfig, true, util.LongTimeout)
+		util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sManagerClient, workerCluster1, true, util.LongTimeout)
+		util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sManagerClient, workerCluster2, true, util.LongTimeout)
 
 		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sManagerClient, managerNs)
 		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sWorker1Client, worker1Ns)
@@ -855,21 +855,25 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 		})
 	})
 
-	ginkgo.When("MultiKueue with Incremental mode", ginkgo.Ordered, func() {
+	ginkgo.When("Incremental mode", ginkgo.Ordered, func() {
+		var defaultManagerKueueCfg *kueueconfig.Configuration
+
 		ginkgo.BeforeAll(func() {
 			ginkgo.By("setting MultiKueue Dispatcher to Incremental", func() {
-				util.UpdateKueueConfiguration(ctx, k8sManagerClient, defaultManagerKueueCfg, managerClusterName, func(cfg *kueueconfig.Configuration) {
-					cfg.MultiKueue = &kueueconfig.MultiKueue{}
+				defaultManagerKueueCfg = util.GetKueueConfiguration(ctx, k8sManagerClient)
+				newCfg := defaultManagerKueueCfg.DeepCopy()
+				util.UpdateKueueConfiguration(ctx, k8sManagerClient, newCfg, managerClusterName, func(cfg *kueueconfig.Configuration) {
+					if cfg.MultiKueue == nil {
+						cfg.MultiKueue = &kueueconfig.MultiKueue{}
+					}
 					cfg.MultiKueue.DispatcherName = ptr.To(kueueconfig.MultiKueueDispatcherModeIncremental)
 				})
 			})
 		})
 		ginkgo.AfterAll(func() {
 			ginkgo.By("setting MultiKueue Dispatcher back to AllAtOnce", func() {
-				util.UpdateKueueConfiguration(ctx, k8sManagerClient, defaultManagerKueueCfg, managerClusterName, func(cfg *kueueconfig.Configuration) {
-					cfg.MultiKueue = &kueueconfig.MultiKueue{}
-					cfg.MultiKueue.DispatcherName = ptr.To(string(kueueconfig.MultiKueueDispatcherModeAllAtOnce))
-				})
+				util.ApplyKueueConfiguration(ctx, k8sManagerClient, defaultManagerKueueCfg)
+				util.RestartKueueController(ctx, k8sManagerClient, managerClusterName)
 			})
 		})
 		ginkgo.It("Should run a job on worker if admitted", func() {
diff --git a/test/e2e/multikueue/suite_test.go b/test/e2e/multikueue/suite_test.go
index 518e97e6e..03699efd0 100644
--- a/test/e2e/multikueue/suite_test.go
+++ b/test/e2e/multikueue/suite_test.go
@@ -43,7 +43,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	jobset "sigs.k8s.io/jobset/api/jobset/v1alpha2"
 
-	"sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/util/kubeversion"
 	"sigs.k8s.io/kueue/test/util"
@@ -68,8 +67,6 @@ var (
 	managerRestClient *rest.RESTClient
 	worker1RestClient *rest.RESTClient
 	worker2RestClient *rest.RESTClient
-
-	defaultManagerKueueCfg *v1beta1.Configuration
 )
 
 func policyRule(group, resource string, verbs ...string) rbacv1.PolicyRule {
@@ -320,8 +317,6 @@ var _ = ginkgo.BeforeSuite(func() {
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 	managerK8SVersion, err = kubeversion.FetchServerVersion(discoveryClient)
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
-
-	defaultManagerKueueCfg = util.GetKueueConfiguration(ctx, k8sManagerClient)
 })
 
 var _ = ginkgo.AfterSuite(func() {
@@ -330,7 +325,4 @@ var _ = ginkgo.AfterSuite(func() {
 
 	gomega.Expect(cleanMultiKueueSecret(ctx, k8sManagerClient, kueueNS, "multikueue1")).To(gomega.Succeed())
 	gomega.Expect(cleanMultiKueueSecret(ctx, k8sManagerClient, kueueNS, "multikueue2")).To(gomega.Succeed())
-
-	util.ApplyKueueConfiguration(ctx, k8sManagerClient, defaultManagerKueueCfg)
-	util.RestartKueueController(ctx, k8sManagerClient, managerClusterName)
 })
diff --git a/test/integration/singlecluster/controller/core/workload_controller_test.go b/test/integration/singlecluster/controller/core/workload_controller_test.go
index 0b6c06f10..47e01b4c6 100644
--- a/test/integration/singlecluster/controller/core/workload_controller_test.go
+++ b/test/integration/singlecluster/controller/core/workload_controller_test.go
@@ -311,7 +311,7 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
 					g.Expect(k8sClient.Get(ctx, wlKey, updatedWl)).To(gomega.Succeed())
 					g.Expect(workload.IsEvictedByDeactivation(updatedWl)).To(gomega.BeTrue())
 					util.ExpectEvictedWorkloadsTotalMetric(clusterQueue.Name, "DeactivatedDueToAdmissionCheck", 1)
-					util.ExpectEvictedWorkloadsOnceTotalMetric(clusterQueue.Name, kueue.WorkloadDeactivated, "", 1)
+					util.ExpectEvictedWorkloadsOnceTotalMetric(clusterQueue.Name, "DeactivatedDueToAdmissionCheck", "", 1)
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 		})
@@ -396,7 +396,7 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
 
 					g.Expect(workload.IsEvictedByDeactivation(updatedWl)).To(gomega.BeTrue())
 					util.ExpectEvictedWorkloadsTotalMetric(clusterQueue.Name, "DeactivatedDueToAdmissionCheck", 1)
-					util.ExpectEvictedWorkloadsOnceTotalMetric(clusterQueue.Name, kueue.WorkloadDeactivated, "", 1)
+					util.ExpectEvictedWorkloadsOnceTotalMetric(clusterQueue.Name, "DeactivatedDueToAdmissionCheck", "", 1)
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 		})
diff --git a/test/integration/singlecluster/scheduler/scheduler_test.go b/test/integration/singlecluster/scheduler/scheduler_test.go
index 71eee5088..9848bb864 100644
--- a/test/integration/singlecluster/scheduler/scheduler_test.go
+++ b/test/integration/singlecluster/scheduler/scheduler_test.go
@@ -2067,6 +2067,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 			util.ExpectEvictedWorkloadsTotalMetric(cq.Name, kueue.WorkloadEvictedByLocalQueueStopped, 1)
+			util.ExpectEvictedWorkloadsOnceTotalMetric(cq.Name, kueue.WorkloadEvictedByLocalQueueStopped, "", 1)
 
 			createdWl := kueue.Workload{}
 			ginkgo.By("Checking the condition of workload is evicted")
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