Repository: kubernetes-sigs/kueue

Issue:
[Flaky E2E] MultiKueue with Incremental mode Should run a job on worker if admitted
<!-- 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/
-->

/kind flake

**What happened**:

End To End MultiKueue Suite: kindest/node:v1.33.1: [It] MultiKueue with Incremental mode Should run a job on worker if admitted 

```
{Expected success, but got an error:
    <*errors.StatusError | 0xc000b91900>: 
    Internal error occurred: failed calling webhook "mresourceflavor.kb.io": failed to call webhook: Post "https://kueue-webhook-service.kueue-system.svc:443/mutate-kueue-x-k8s-io-v1beta1-resourceflavor?timeout=10s": write tcp 172.18.0.7:59888->10.244.1.7:9443: write: connection reset by peer
    {
        ErrStatus: {
            TypeMeta: {Kind: "", APIVersion: ""},
            ListMeta: {
                SelfLink: "",
                ResourceVersion: "",
                Continue: "",
                RemainingItemCount: nil,
            },
            Status: "Failure",
            Message: "Internal error occurred: failed calling webhook \"mresourceflavor.kb.io\": failed to call webhook: Post \"https://kueue-webhook-service.kueue-system.svc:443/mutate-kueue-x-k8s-io-v1beta1-resourceflavor?timeout=10s\": write tcp 172.18.0.7:59888->10.244.1.7:9443: write: connection reset by peer",
            Reason: "InternalError",
            Details: {
                Name: "",
                Group: "",
                Kind: "",
                UID: "",
                Causes: [
                    {
                        Type: "",
                        Message: "failed calling webhook \"mresourceflavor.kb.io\": failed to call webhook: Post \"https://kueue-webhook-service.kueue-system.svc:443/mutate-kueue-x-k8s-io-v1beta1-resourceflavor?timeout=10s\": write tcp 172.18.0.7:59888->10.244.1.7:9443: write: connection reset by peer",
                        Field: "",
                    },
                ],
                RetryAfterSeconds: 0,
            },
            Code: 500,
        },
    } failed [FAILED] Expected success, but got an error:
    <*errors.StatusError | 0xc000b91900>: 
    Internal error occurred: failed calling webhook "mresourceflavor.kb.io": failed to call webhook: Post "https://kueue-webhook-service.kueue-system.svc:443/mutate-kueue-x-k8s-io-v1beta1-resourceflavor?timeout=10s": write tcp 172.18.0.7:59888->10.244.1.7:9443: write: connection reset by peer
    {
        ErrStatus: {
            TypeMeta: {Kind: "", APIVersion: ""},
            ListMeta: {
                SelfLink: "",
                ResourceVersion: "",
                Continue: "",
                RemainingItemCount: nil,
            },
            Status: "Failure",
            Message: "Internal error occurred: failed calling webhook \"mresourceflavor.kb.io\": failed to call webhook: Post \"https://kueue-webhook-service.kueue-system.svc:443/mutate-kueue-x-k8s-io-v1beta1-resourceflavor?timeout=10s\": write tcp 172.18.0.7:59888->10.244.1.7:9443: write: connection reset by peer",
            Reason: "InternalError",
            Details: {
                Name: "",
                Group: "",
                Kind: "",
                UID: "",
                Causes: [
                    {
                        Type: "",
                        Message: "failed calling webhook \"mresourceflavor.kb.io\": failed to call webhook: Post \"https://kueue-webhook-service.kueue-system.svc:443/mutate-kueue-x-k8s-io-v1beta1-resourceflavor?timeout=10s\": write tcp 172.18.0.7:59888->10.244.1.7:9443: write: connection reset by peer",
                        Field: "",
                    },
                ],
                RetryAfterSeconds: 0,
            },
            Code: 500,
        },
    }
In [BeforeAll] at: /home/prow/go/src/sigs.k8s.io/kueue/test/e2e/multikueue/e2e_test.go:1031 @ 07/29/25 09:05:11.601
}
```

**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-e2e-multikueue-main/1950115850333720576

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

**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:
@mbobrovskyi: /cc @mszadkow @mimowo: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6287/pull-kueue-test-e2e-multikueue-main/1950551855390003200@mimowo: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6284/pull-kueue-test-e2e-multikueue-main/1950537174688468992@mszadkow: /assign
@mimowo: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6297/pull-kueue-test-e2e-multikueue-main/1950828015646674944@mimowo: seems like the most common flake currently
@mszadkow: Yes, there is nothing wrong with the test itself, I am 99% positive about that.
I have checked all the logs etc. every time it's the same pattern with the webhook, before the test even really starts.
However as @mbobrovskyi mentioned the other test might ruin this one's performance - 
`The connection to a worker cluster is unreliable`.
I try to move this test in the hierarchy to eliminate flakiness, but `The connection to a worker cluster is unreliable` is the one that has to be fixed.@mimowo: > I try to move this test in the hierarchy to eliminate flakiness, but The connection to a worker cluster is unreliable is the one that has to be fixed.

Because the "The connection to a worker cluster is unreliable" test breaks the connectivity between the clusters? This could make sense. If this is the case then indeed it is responsibility of this test to restore the connection and test it is ok.@mimowo: OTOH, it does not quite make sense, the test disrupts connectivity between clusters, not within a cluster. The webhook failures suggest requests failing *within* the cluster. I don't think we would directly call webhooks from external cluster.

Given how common the flake is I wouldn't assume it is network issue within cluster (these are super rare). The most likely explanation for such errors is that Kueue (which hosts the webhooks) is non-responsible (for example restarting or crashing).
@mimowo: still flake https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6329/pull-kueue-test-e2e-multikueue-release-0-13/1950934478683639808
@mbobrovskyi: One more https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/6345/pull-kueue-test-e2e-multikueue-main/1951172197842161664

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/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 09db8ea29..fbbcfb61e 100644
--- a/pkg/controller/tas/node_failure_controller.go
+++ b/pkg/controller/tas/node_failure_controller.go
@@ -300,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) {
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/preemption.go b/pkg/scheduler/preemption/preemption.go
index 978dd9d01..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 {
diff --git a/site/content/en/docs/concepts/cluster_queue.md b/site/content/en/docs/concepts/cluster_queue.md
index 4be5b3c1f..75360301a 100644
--- a/site/content/en/docs/concepts/cluster_queue.md
+++ b/site/content/en/docs/concepts/cluster_queue.md
@@ -475,12 +475,14 @@ the heuristics that Kueue implements to preempt as few Workloads as possible.
 
 ## FlavorFungibility
 
-When there is not enough nominal quota of resources in a ResourceFlavor, the incoming Workload can borrow
-quota or preempt running Workloads in the ClusterQueue or Cohort.
-
-Kueue evaluates the flavors in a ClusterQueue in order. You can influence whether to prioritize
-preemptions or borrowing in a flavor before trying to accommodate the Workload in the next flavor, by
-setting the `flavorFungibility` field.
+When there is not enough nominal quota of resources in a ResourceFlavor, the
+incoming Workload can borrow quota or preempt running Workloads in the
+ClusterQueue or Cohort. In the ClusterQueue, there can be multiple Flavors
+for each resource, with potentially different amounts of quota available in each.
+Kueue evaluates the Flavors in a ClusterQueue in order and for each it
+checks if the required resource of the workload fits in this ResourceFlavor and
+if borrowing or preemption is required. You can influence the Flavor selection
+process by configuring the `flavorFungibility` field.
 
 A configuration for a ClusterQueue that configures this behavior looks like the following:
 
@@ -497,17 +499,40 @@ spec:
 
 The fields above do the following:
 
-- `whenCanBorrow` determines whether a workload should stop finding a better assignment if it can get enough resource by borrowing in current ResourceFlavor. The possible values are:
-  - `Borrow` (default): ClusterQueue stops finding a better assignment.
-  - `TryNextFlavor`: ClusterQueue tries the next ResourceFlavor to see if the workload can get a better assignment.
-- `whenCanPreempt` determines whether a workload should try preemption in current ResourceFlavor before try the next one. The possible values are:
-  - `Preempt`: ClusterQueue stops trying preemption in current ResourceFlavor and starts from the next one if preempting failed.
-  - `TryNextFlavor` (default): ClusterQueue tries the next ResourceFlavor to see if the workload can fit in the ResourceFlavor.
+- `whenCanBorrow` defines what should happen if a workload can get enough resource by
+borrowing in current ResourceFlavor. The possible values are:
+  - `Borrow` (default): Kueue stops looking for a better assignment.
+  - `TryNextFlavor`: Kueue tries the next ResourceFlavor.
+- `whenCanPreempt` defines what should happen if a workload can get enough resource by
+preempting in current ResourceFlavor. The possible values are:
+  - `Preempt`: Kueue stops looking for a better assignment.
+  - `TryNextFlavor` (default): Kueue tries the next ResourceFlavor.
+
+If during the search, Kueue finds some ResourceFlavor in which it can fit
+without preemption or borrowing, such ResourceFlavor is immediately selected,
+regardless of the above configuration. Otherwise, out of the considered
+ResourceFlavors, Kueue selects a one that fits the workload using borrowing
+(without preemptions). If there is no such ResourceFlavor, Kueue selects a Flavor
+that uses preemption and is preferably not borrowing.
+
+By default Kueue avoids preemptions and prefers borrowing when assigning Flavors.
+Borrowing is not disruptive to other workloads but a
+workload that borrows risks being prempted (since it is using nominal quota
+from some other Cluster Queue). If you prefer to preempt rather than borrow when possible,
+you can enable the feature gate `FlavorFungibilityImplicitPreferenceDefault`, which
+changes the default preference as follows: If `.spec.flavorFungibility.whenCanBorrow` is `TryNextFlavor`,
+it assumes that preemption is preferred over borrowing and otherwise it assumes
+that borrowing is preferred over preemption.
+
+{{% alert title="Note" color="primary" %}}
+`FlavorFungibilityImplicitPreferenceDefault` is currently an alpha feature,
+introduced to Kueue in version 0.13 and it is not enabled by default.
 
-By default, the incoming workload stops trying the next flavor if the workload can get enough borrowed resources.
-And Kueue triggers preemption only after Kueue determines that the remaining ResourceFlavors can't fit the workload.
+To enable the feature, you have to set the `FlavorFungibilityImplicitPreferenceDefault`
+feature gate to `true`. Check the [Installation](/docs/installation/#change-the-feature-gates-configuration)
+guide for details on feature gate configuration.
+{{% /alert %}}
 
-Note that, whenever possible and when the configured policy allows it, Kueue avoids preemptions if it can fit a Workload by borrowing.
 
 ## StopPolicy
 
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/test/e2e/multikueue/customconfigs_test.go b/test/e2e/multikueue/customconfigs_test.go
new file mode 100644
index 000000000..394334a19
--- /dev/null
+++ b/test/e2e/multikueue/customconfigs_test.go
@@ -0,0 +1,262 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mke2e
+
+import (
+	"fmt"
+
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kueueconfig "sigs.k8s.io/kueue/apis/config/v1beta1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
+	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
+	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var _ = ginkgo.Describe("MultiKueue with custom configs", func() {
+	var (
+		managerNs *corev1.Namespace
+		worker1Ns *corev1.Namespace
+		worker2Ns *corev1.Namespace
+
+		workerCluster1   *kueue.MultiKueueCluster
+		workerCluster2   *kueue.MultiKueueCluster
+		multiKueueConfig *kueue.MultiKueueConfig
+		multiKueueAc     *kueue.AdmissionCheck
+		managerFlavor    *kueue.ResourceFlavor
+		managerCq        *kueue.ClusterQueue
+		managerLq        *kueue.LocalQueue
+
+		worker1Flavor *kueue.ResourceFlavor
+		worker1Cq     *kueue.ClusterQueue
+		worker1Lq     *kueue.LocalQueue
+
+		worker2Flavor *kueue.ResourceFlavor
+		worker2Cq     *kueue.ClusterQueue
+		worker2Lq     *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeEach(func() {
+		managerNs = util.CreateNamespaceFromPrefixWithLog(ctx, k8sManagerClient, "multikueue-")
+		worker1Ns = util.CreateNamespaceWithLog(ctx, k8sWorker1Client, managerNs.Name)
+		worker2Ns = util.CreateNamespaceWithLog(ctx, k8sWorker2Client, managerNs.Name)
+
+		workerCluster1 = utiltesting.MakeMultiKueueCluster("worker1").KubeConfig(kueue.SecretLocationType, "multikueue1").Obj()
+		util.MustCreate(ctx, k8sManagerClient, workerCluster1)
+
+		workerCluster2 = utiltesting.MakeMultiKueueCluster("worker2").KubeConfig(kueue.SecretLocationType, "multikueue2").Obj()
+		util.MustCreate(ctx, k8sManagerClient, workerCluster2)
+
+		multiKueueConfig = utiltesting.MakeMultiKueueConfig("multikueueconfig").Clusters("worker1", "worker2").Obj()
+		util.MustCreate(ctx, k8sManagerClient, multiKueueConfig)
+
+		multiKueueAc = utiltesting.MakeAdmissionCheck("ac1").
+			ControllerName(kueue.MultiKueueControllerName).
+			Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", multiKueueConfig.Name).
+			Obj()
+		util.MustCreate(ctx, k8sManagerClient, multiKueueAc)
+
+		ginkgo.By("wait for check active", func() {
+			updatedAc := kueue.AdmissionCheck{}
+			acKey := client.ObjectKeyFromObject(multiKueueAc)
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sManagerClient.Get(ctx, acKey, &updatedAc)).To(gomega.Succeed())
+				g.Expect(updatedAc.Status.Conditions).To(utiltesting.HaveConditionStatusTrue(kueue.AdmissionCheckActive))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+		managerFlavor = utiltesting.MakeResourceFlavor("default").Obj()
+		util.MustCreate(ctx, k8sManagerClient, managerFlavor)
+
+		managerCq = utiltesting.MakeClusterQueue("q1").
+			ResourceGroup(
+				*utiltesting.MakeFlavorQuotas(managerFlavor.Name).
+					Resource(corev1.ResourceCPU, "2").
+					Resource(corev1.ResourceMemory, "2G").
+					Obj(),
+			).
+			AdmissionChecks(kueue.AdmissionCheckReference(multiKueueAc.Name)).
+			Obj()
+		util.MustCreate(ctx, k8sManagerClient, managerCq)
+
+		managerLq = utiltesting.MakeLocalQueue(managerCq.Name, managerNs.Name).ClusterQueue(managerCq.Name).Obj()
+		util.MustCreate(ctx, k8sManagerClient, managerLq)
+
+		worker1Flavor = utiltesting.MakeResourceFlavor("default").Obj()
+		util.MustCreate(ctx, k8sWorker1Client, worker1Flavor)
+
+		worker1Cq = utiltesting.MakeClusterQueue("q1").
+			ResourceGroup(
+				*utiltesting.MakeFlavorQuotas(worker1Flavor.Name).
+					Resource(corev1.ResourceCPU, "2").
+					Resource(corev1.ResourceMemory, "1G").
+					Obj(),
+			).
+			Obj()
+		util.MustCreate(ctx, k8sWorker1Client, worker1Cq)
+
+		worker1Lq = utiltesting.MakeLocalQueue(worker1Cq.Name, worker1Ns.Name).ClusterQueue(worker1Cq.Name).Obj()
+		util.MustCreate(ctx, k8sWorker1Client, worker1Lq)
+
+		worker2Flavor = utiltesting.MakeResourceFlavor("default").Obj()
+		util.MustCreate(ctx, k8sWorker2Client, worker2Flavor)
+
+		worker2Cq = utiltesting.MakeClusterQueue("q1").
+			ResourceGroup(
+				*utiltesting.MakeFlavorQuotas(worker2Flavor.Name).
+					Resource(corev1.ResourceCPU, "1").
+					Resource(corev1.ResourceMemory, "2G").
+					Obj(),
+			).
+			Obj()
+		util.MustCreate(ctx, k8sWorker2Client, worker2Cq)
+
+		worker2Lq = utiltesting.MakeLocalQueue(worker2Cq.Name, worker2Ns.Name).ClusterQueue(worker2Cq.Name).Obj()
+		util.MustCreate(ctx, k8sWorker2Client, worker2Lq)
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sManagerClient, managerNs)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(ctx, k8sWorker1Client, worker1Ns)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(ctx, k8sWorker2Client, worker2Ns)).To(gomega.Succeed())
+
+		util.ExpectObjectToBeDeleted(ctx, k8sWorker1Client, worker1Cq, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sWorker1Client, worker1Flavor, true)
+
+		util.ExpectObjectToBeDeleted(ctx, k8sWorker2Client, worker2Cq, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sWorker2Client, worker2Flavor, 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)
+		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sWorker2Client, worker2Ns)
+	})
+
+	ginkgo.When("Dispatcher Incremental mode", ginkgo.Ordered, func() {
+		var defaultManagerKueueCfg *kueueconfig.Configuration
+
+		ginkgo.BeforeAll(func() {
+			ginkgo.By("setting MultiKueue Dispatcher to Incremental", func() {
+				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.ApplyKueueConfiguration(ctx, k8sManagerClient, defaultManagerKueueCfg)
+				util.RestartKueueController(ctx, k8sManagerClient, managerClusterName)
+			})
+		})
+		ginkgo.It("Should run a job on worker if admitted", func() {
+			// Since it requires 2G of memory, this job can only be admitted in worker 2.
+			job := testingjob.MakeJob("job", managerNs.Name).
+				Queue(kueue.LocalQueueName(managerLq.Name)).
+				RequestAndLimit(corev1.ResourceCPU, "1").
+				RequestAndLimit(corev1.ResourceMemory, "2G").
+				TerminationGracePeriod(1).
+				// Give it the time to be observed Active in the live status update step.
+				Image(util.GetAgnHostImage(), util.BehaviorWaitForDeletion).
+				Obj()
+
+			ginkgo.By("Creating the job", func() {
+				util.MustCreate(ctx, k8sManagerClient, job)
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdJob := &batchv1.Job{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
+					g.Expect(ptr.Deref(createdJob.Spec.ManagedBy, "")).To(gomega.BeEquivalentTo(kueue.MultiKueueControllerName))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+			createdLeaderWorkload := &kueue.Workload{}
+			wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: managerNs.Name}
+			// the execution should be given to the worker
+			ginkgo.By("Waiting to be admitted in worker2, and the manager's job unsuspended", 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 "worker2"`,
+					}, cmpopts.IgnoreFields(kueue.AdmissionCheckState{}, "LastTransitionTime")))
+					g.Expect(ptr.Deref(createdLeaderWorkload.Status.ClusterName, "")).To(gomega.Equal(workerCluster2.Name))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdJob := &batchv1.Job{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
+					g.Expect(ptr.Deref(createdJob.Spec.Suspend, false)).To(gomega.BeFalse())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Waiting for the job to get status updates", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdJob := batchv1.Job{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), &createdJob)).To(gomega.Succeed())
+					g.Expect(createdJob.Status.StartTime).NotTo(gomega.BeNil())
+					g.Expect(createdJob.Status.Active).To(gomega.Equal(int32(1)))
+					g.Expect(createdJob.Status.CompletionTime).To(gomega.BeNil())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Finishing the job's pod", func() {
+				listOpts := util.GetListOptsFromLabel(fmt.Sprintf("batch.kubernetes.io/job-name=%s", job.Name))
+				util.WaitForActivePodsAndTerminate(ctx, k8sWorker2Client, worker2RestClient, worker2Cfg, job.Namespace, 1, 0, listOpts)
+			})
+
+			ginkgo.By("Waiting for the job to finish", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sManagerClient.Get(ctx, wlLookupKey, createdLeaderWorkload)).To(gomega.Succeed())
+					g.Expect(createdLeaderWorkload.Status.Conditions).To(utiltesting.HaveConditionStatusTrueAndReason(kueue.WorkloadFinished, kueue.WorkloadFinishedReasonSucceeded))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking no objects are left in the worker clusters and the job is completed", func() {
+				expectObjectToBeDeletedOnWorkerClusters(ctx, createdLeaderWorkload)
+				expectObjectToBeDeletedOnWorkerClusters(ctx, job)
+
+				createdJob := &batchv1.Job{}
+				gomega.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
+				gomega.Expect(createdJob.Status.Conditions).To(gomega.ContainElement(gomega.BeComparableTo(
+					batchv1.JobCondition{
+						Type:   batchv1.JobComplete,
+						Status: corev1.ConditionTrue,
+					},
+					cmpopts.IgnoreFields(batchv1.JobCondition{}, "LastTransitionTime", "LastProbeTime", "Reason", "Message"))))
+			})
+		})
+	})
+})
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index f85e849c4..628b9e4d6 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -39,7 +39,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	jobset "sigs.k8s.io/jobset/api/jobset/v1alpha2"
 
-	kueueconfig "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	workloadaw "sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
@@ -855,105 +854,6 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 		})
 	})
 
-	ginkgo.When("Incremental mode", ginkgo.Ordered, func() {
-		var defaultManagerKueueCfg *kueueconfig.Configuration
-
-		ginkgo.BeforeAll(func() {
-			ginkgo.By("setting MultiKueue Dispatcher to Incremental", func() {
-				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.ApplyKueueConfiguration(ctx, k8sManagerClient, defaultManagerKueueCfg)
-				util.RestartKueueController(ctx, k8sManagerClient, managerClusterName)
-			})
-		})
-		ginkgo.It("Should run a job on worker if admitted", func() {
-			// Since it requires 2G of memory, this job can only be admitted in worker 2.
-			job := testingjob.MakeJob("job", managerNs.Name).
-				Queue(kueue.LocalQueueName(managerLq.Name)).
-				RequestAndLimit(corev1.ResourceCPU, "1").
-				RequestAndLimit(corev1.ResourceMemory, "2G").
-				TerminationGracePeriod(1).
-				// Give it the time to be observed Active in the live status update step.
-				Image(util.GetAgnHostImage(), util.BehaviorWaitForDeletion).
-				Obj()
-
-			ginkgo.By("Creating the job", func() {
-				util.MustCreate(ctx, k8sManagerClient, job)
-				gomega.Eventually(func(g gomega.Gomega) {
-					createdJob := &batchv1.Job{}
-					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
-					g.Expect(ptr.Deref(createdJob.Spec.ManagedBy, "")).To(gomega.BeEquivalentTo(kueue.MultiKueueControllerName))
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
-			})
-			createdLeaderWorkload := &kueue.Workload{}
-			wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: managerNs.Name}
-			// the execution should be given to the worker
-			ginkgo.By("Waiting to be admitted in worker2, and the manager's job unsuspended", 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 "worker2"`,
-					}, cmpopts.IgnoreFields(kueue.AdmissionCheckState{}, "LastTransitionTime")))
-					g.Expect(ptr.Deref(createdLeaderWorkload.Status.ClusterName, "")).To(gomega.Equal(workerCluster2.Name))
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
-
-				gomega.Eventually(func(g gomega.Gomega) {
-					createdJob := &batchv1.Job{}
-					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
-					g.Expect(ptr.Deref(createdJob.Spec.Suspend, false)).To(gomega.BeFalse())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Waiting for the job to get status updates", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					createdJob := batchv1.Job{}
-					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), &createdJob)).To(gomega.Succeed())
-					g.Expect(createdJob.Status.StartTime).NotTo(gomega.BeNil())
-					g.Expect(createdJob.Status.Active).To(gomega.Equal(int32(1)))
-					g.Expect(createdJob.Status.CompletionTime).To(gomega.BeNil())
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Finishing the job's pod", func() {
-				listOpts := util.GetListOptsFromLabel(fmt.Sprintf("batch.kubernetes.io/job-name=%s", job.Name))
-				util.WaitForActivePodsAndTerminate(ctx, k8sWorker2Client, worker2RestClient, worker2Cfg, job.Namespace, 1, 0, listOpts)
-			})
-
-			ginkgo.By("Waiting for the job to finish", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sManagerClient.Get(ctx, wlLookupKey, createdLeaderWorkload)).To(gomega.Succeed())
-					g.Expect(createdLeaderWorkload.Status.Conditions).To(utiltesting.HaveConditionStatusTrueAndReason(kueue.WorkloadFinished, kueue.WorkloadFinishedReasonSucceeded))
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Checking no objects are left in the worker clusters and the job is completed", func() {
-				expectObjectToBeDeletedOnWorkerClusters(ctx, createdLeaderWorkload)
-				expectObjectToBeDeletedOnWorkerClusters(ctx, job)
-
-				createdJob := &batchv1.Job{}
-				gomega.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
-				gomega.Expect(createdJob.Status.Conditions).To(gomega.ContainElement(gomega.BeComparableTo(
-					batchv1.JobCondition{
-						Type:   batchv1.JobComplete,
-						Status: corev1.ConditionTrue,
-					},
-					cmpopts.IgnoreFields(batchv1.JobCondition{}, "LastTransitionTime", "LastProbeTime", "Reason", "Message"))))
-			})
-		})
-	})
-
 	ginkgo.When("The connection to a worker cluster is unreliable", func() {
 		ginkgo.It("Should update the cluster status to reflect the connection state", func() {
 			worker1Cq2 := utiltesting.MakeClusterQueue("q2").
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")