Repository: kubernetes-sigs/kueue

Issue:
MK Dispatcher: Test for possible race condition on AC and ClusterName set
**What would you like to be added**:

As mentioned [here](https://github.com/kubernetes-sigs/kueue/pull/5782#discussion_r2230404869)
We need to test and possibly fix the race condition that could happen:

>  when the nominatedClusterNames is changed preventing setting clusterName, but not preventing setting AC as Ready, and admitting the workload

**Why is this needed**:

**Completion requirements**:

This enhancement requires the following artifacts:

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

The artifacts should be linked in subsequent comments.


Discussion:
@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle rotten

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@mimowo: /close
To avoid distractions. The code has evolved a lot since then, and I'm not even sure I was right back then. I can see currently we set the AC=Ready and the status.clusterName in the same request, so I don't expect a race here: https://github.com/kubernetes-sigs/kueue/blob/main/pkg/controller/admissionchecks/multikueue/workload.go#L639-L648@k8s-ci-robot: @mimowo: Closing this issue.

<details>

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

>/close
>To avoid distractions. The code has evolved a lot since then, and I'm not even sure I was right back then. I can see currently we set the AC=Ready and the status.clusterName in the same request, so I don't expect a race here: https://github.com/kubernetes-sigs/kueue/blob/main/pkg/controller/admissionchecks/multikueue/workload.go#L639-L648


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

Solution patch:
diff --git a/apis/config/v1beta1/configuration_types.go b/apis/config/v1beta1/configuration_types.go
index f8705932a..bbd083f39 100644
--- a/apis/config/v1beta1/configuration_types.go
+++ b/apis/config/v1beta1/configuration_types.go
@@ -273,8 +273,24 @@ type MultiKueue struct {
 	// Defaults to 15 minutes.
 	// +optional
 	WorkerLostTimeout *metav1.Duration `json:"workerLostTimeout,omitempty"`
+
+	// DispatcherName defines the dispatcher responsible for selecting worker clusters to handle the workload.
+	// - If specified, the workload will be handled by the named dispatcher.
+	// - If not specified, the workload will be handled by the default ("kueue.x-k8s.io/multikueue-dispatcher-all-at-once") dispatcher.
+	// +optional
+	DispatcherName *string `json:"dispatcherName,omitempty"`
 }
 
+const (
+	// MultiKueueDispatcherModeAllAtOnce is the name of dispatcher mode where all worker clusters are considered at once
+	// and the first one accepting the workload is selected.
+	MultiKueueDispatcherModeAllAtOnce = "kueue.x-k8s.io/multikueue-dispatcher-all-at-once"
+
+	// MultiKueueDispatcherModeIncremental is the name of dispatcher mode where worker clusters are incrementally added to the pool of nominated clusters.
+	// The process begins with up to 3 initial clusters and expands the pool by up to 3 clusters at a time (if fewer remain, all are added).
+	MultiKueueDispatcherModeIncremental = "kueue.x-k8s.io/multikueue-dispatcher-incremental"
+)
+
 type RequeuingStrategy struct {
 	// Timestamp defines the timestamp used for re-queuing a Workload
 	// that was evicted due to Pod readiness. The possible values are:
diff --git a/apis/config/v1beta1/defaults.go b/apis/config/v1beta1/defaults.go
index 2c7792c8f..eef291329 100644
--- a/apis/config/v1beta1/defaults.go
+++ b/apis/config/v1beta1/defaults.go
@@ -129,6 +129,7 @@ func SetDefaults_Configuration(cfg *Configuration) {
 	cfg.MultiKueue.GCInterval = cmp.Or(cfg.MultiKueue.GCInterval, &metav1.Duration{Duration: DefaultMultiKueueGCInterval})
 	cfg.MultiKueue.Origin = ptr.To(cmp.Or(ptr.Deref(cfg.MultiKueue.Origin, ""), DefaultMultiKueueOrigin))
 	cfg.MultiKueue.WorkerLostTimeout = cmp.Or(cfg.MultiKueue.WorkerLostTimeout, &metav1.Duration{Duration: DefaultMultiKueueWorkerLostTimeout})
+	cfg.MultiKueue.DispatcherName = cmp.Or(cfg.MultiKueue.DispatcherName, ptr.To(MultiKueueDispatcherModeAllAtOnce))
 
 	if fs := cfg.FairSharing; fs != nil && fs.Enable && len(fs.PreemptionStrategies) == 0 {
 		fs.PreemptionStrategies = []PreemptionStrategy{LessThanOrEqualToFinalShare, LessThanInitialShare}
diff --git a/apis/config/v1beta1/zz_generated.deepcopy.go b/apis/config/v1beta1/zz_generated.deepcopy.go
index 293cb2c28..3a561c392 100644
--- a/apis/config/v1beta1/zz_generated.deepcopy.go
+++ b/apis/config/v1beta1/zz_generated.deepcopy.go
@@ -392,6 +392,11 @@ func (in *MultiKueue) DeepCopyInto(out *MultiKueue) {
 		*out = new(v1.Duration)
 		**out = **in
 	}
+	if in.DispatcherName != nil {
+		in, out := &in.DispatcherName, &out.DispatcherName
+		*out = new(string)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MultiKueue.
diff --git a/apis/kueue/v1beta1/workload_types.go b/apis/kueue/v1beta1/workload_types.go
index 0255d53ab..ce775279b 100644
--- a/apis/kueue/v1beta1/workload_types.go
+++ b/apis/kueue/v1beta1/workload_types.go
@@ -344,6 +344,8 @@ type PodSet struct {
 }
 
 // WorkloadStatus defines the observed state of Workload
+// +kubebuilder:validation:XValidation:rule="!has(oldSelf.clusterName) || !has(self.clusterName) || oldSelf.clusterName == self.clusterName", message="clusterName is immutable once set"
+// +kubebuilder:validation:XValidation:rule="!has(self.clusterName) || (!has(self.nominatedClusterNames) || (has(self.nominatedClusterNames) && size(self.nominatedClusterNames) == 0))", message="clusterName and nominatedClusterNames are mutually exclusive"
 type WorkloadStatus struct {
 	// admission holds the parameters of the admission of the workload by a
 	// ClusterQueue. admission can be set back to null, but its fields cannot be
@@ -411,6 +413,21 @@ type WorkloadStatus struct {
 	//
 	// +optional
 	SchedulingStats *SchedulingStats `json:"schedulingStats,omitempty"`
+
+	// nominatedClusterNames specifies the list of cluster names that have been nominated for scheduling.
+	// This field is mutually exclusive with the `.status.clusterName` field, and is reset when
+	// `status.clusterName` is set.
+	// This field is optional.
+	//
+	// +listType=atomic
+	// +kubebuilder:validation:MaxItems=10
+	// +optional
+	NominatedClusterNames []string `json:"nominatedClusterNames,omitempty"`
+
+	// clusterName is the name of the cluster where the workload is actually assigned.
+	// This field is reset after the Workload is evicted.
+	// +optional
+	ClusterName *string `json:"clusterName,omitempty"`
 }
 
 type SchedulingStats struct {
diff --git a/apis/kueue/v1beta1/zz_generated.deepcopy.go b/apis/kueue/v1beta1/zz_generated.deepcopy.go
index 3f8e41834..1214c32d6 100644
--- a/apis/kueue/v1beta1/zz_generated.deepcopy.go
+++ b/apis/kueue/v1beta1/zz_generated.deepcopy.go
@@ -1991,6 +1991,16 @@ func (in *WorkloadStatus) DeepCopyInto(out *WorkloadStatus) {
 		*out = new(SchedulingStats)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.NominatedClusterNames != nil {
+		in, out := &in.NominatedClusterNames, &out.NominatedClusterNames
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.ClusterName != nil {
+		in, out := &in.ClusterName, &out.ClusterName
+		*out = new(string)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkloadStatus.
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
index fc58dac34..28800086f 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
@@ -8232,6 +8232,11 @@ spec:
                   x-kubernetes-list-map-keys:
                     - name
                   x-kubernetes-list-type: map
+                clusterName:
+                  description: |-
+                    clusterName is the name of the cluster where the workload is actually assigned.
+                    This field is reset after the Workload is evicted.
+                  type: string
                 conditions:
                   description: |-
                     conditions hold the latest available observations of the Workload
@@ -8300,6 +8305,17 @@ spec:
                   x-kubernetes-list-map-keys:
                     - type
                   x-kubernetes-list-type: map
+                nominatedClusterNames:
+                  description: |-
+                    nominatedClusterNames specifies the list of cluster names that have been nominated for scheduling.
+                    This field is mutually exclusive with the `.status.clusterName` field, and is reset when
+                    `status.clusterName` is set.
+                    This field is optional.
+                  items:
+                    type: string
+                  maxItems: 10
+                  type: array
+                  x-kubernetes-list-type: atomic
                 reclaimablePods:
                   description: |-
                     reclaimablePods keeps track of the number pods within a podset for which
@@ -8416,6 +8432,11 @@ spec:
                       x-kubernetes-list-type: map
                   type: object
               type: object
+              x-kubernetes-validations:
+                - message: clusterName is immutable once set
+                  rule: '!has(oldSelf.clusterName) || !has(self.clusterName) || oldSelf.clusterName == self.clusterName'
+                - message: clusterName and nominatedClusterNames are mutually exclusive
+                  rule: '!has(self.clusterName) || (!has(self.nominatedClusterNames) || (has(self.nominatedClusterNames) && size(self.nominatedClusterNames) == 0))'
           type: object
           x-kubernetes-validations:
             - message: podSetAssignments must have the same number of podSets as the spec
diff --git a/client-go/applyconfiguration/kueue/v1beta1/workloadstatus.go b/client-go/applyconfiguration/kueue/v1beta1/workloadstatus.go
index 1d1239d8e..769f0db21 100644
--- a/client-go/applyconfiguration/kueue/v1beta1/workloadstatus.go
+++ b/client-go/applyconfiguration/kueue/v1beta1/workloadstatus.go
@@ -32,6 +32,8 @@ type WorkloadStatusApplyConfiguration struct {
 	ResourceRequests                     []PodSetRequestApplyConfiguration       `json:"resourceRequests,omitempty"`
 	AccumulatedPastExexcutionTimeSeconds *int32                                  `json:"accumulatedPastExexcutionTimeSeconds,omitempty"`
 	SchedulingStats                      *SchedulingStatsApplyConfiguration      `json:"schedulingStats,omitempty"`
+	NominatedClusterNames                []string                                `json:"nominatedClusterNames,omitempty"`
+	ClusterName                          *string                                 `json:"clusterName,omitempty"`
 }
 
 // WorkloadStatusApplyConfiguration constructs a declarative configuration of the WorkloadStatus type for use with
@@ -123,3 +125,21 @@ func (b *WorkloadStatusApplyConfiguration) WithSchedulingStats(value *Scheduling
 	b.SchedulingStats = value
 	return b
 }
+
+// WithNominatedClusterNames adds the given value to the NominatedClusterNames field in the declarative configuration
+// and returns the receiver, so that objects can be build by chaining "With" function invocations.
+// If called multiple times, values provided by each call will be appended to the NominatedClusterNames field.
+func (b *WorkloadStatusApplyConfiguration) WithNominatedClusterNames(values ...string) *WorkloadStatusApplyConfiguration {
+	for i := range values {
+		b.NominatedClusterNames = append(b.NominatedClusterNames, values[i])
+	}
+	return b
+}
+
+// WithClusterName sets the ClusterName field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the ClusterName field is set to the value of the last call.
+func (b *WorkloadStatusApplyConfiguration) WithClusterName(value string) *WorkloadStatusApplyConfiguration {
+	b.ClusterName = &value
+	return b
+}
diff --git a/cmd/kueue/main.go b/cmd/kueue/main.go
index 56496f1c9..1d96477e6 100644
--- a/cmd/kueue/main.go
+++ b/cmd/kueue/main.go
@@ -346,6 +346,7 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *cache.Cache
 			multikueue.WithOrigin(ptr.Deref(cfg.MultiKueue.Origin, configapi.DefaultMultiKueueOrigin)),
 			multikueue.WithWorkerLostTimeout(cfg.MultiKueue.WorkerLostTimeout.Duration),
 			multikueue.WithAdapters(adapters),
+			multikueue.WithDispatcherName(ptr.Deref(cfg.MultiKueue.DispatcherName, configapi.MultiKueueDispatcherModeAllAtOnce)),
 		); err != nil {
 			return fmt.Errorf("could not setup MultiKueue controller: %w", err)
 		}
@@ -357,7 +358,7 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *cache.Cache
 		}
 	}
 
-	if failedWebhook, err := webhooks.Setup(mgr); err != nil {
+	if failedWebhook, err := webhooks.Setup(mgr, ptr.Deref(cfg.MultiKueue.DispatcherName, configapi.MultiKueueDispatcherModeAllAtOnce)); err != nil {
 		return fmt.Errorf("unable to create webhook %s: %w", failedWebhook, err)
 	}
 
diff --git a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
index 4d9ac5719..bbe767756 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
@@ -8713,6 +8713,11 @@ spec:
                 x-kubernetes-list-map-keys:
                 - name
                 x-kubernetes-list-type: map
+              clusterName:
+                description: |-
+                  clusterName is the name of the cluster where the workload is actually assigned.
+                  This field is reset after the Workload is evicted.
+                type: string
               conditions:
                 description: |-
                   conditions hold the latest available observations of the Workload
@@ -8782,6 +8787,17 @@ spec:
                 x-kubernetes-list-map-keys:
                 - type
                 x-kubernetes-list-type: map
+              nominatedClusterNames:
+                description: |-
+                  nominatedClusterNames specifies the list of cluster names that have been nominated for scheduling.
+                  This field is mutually exclusive with the `.status.clusterName` field, and is reset when
+                  `status.clusterName` is set.
+                  This field is optional.
+                items:
+                  type: string
+                maxItems: 10
+                type: array
+                x-kubernetes-list-type: atomic
               reclaimablePods:
                 description: |-
                   reclaimablePods keeps track of the number pods within a podset for which
@@ -8903,6 +8919,14 @@ spec:
                     x-kubernetes-list-type: map
                 type: object
             type: object
+            x-kubernetes-validations:
+            - message: clusterName is immutable once set
+              rule: '!has(oldSelf.clusterName) || !has(self.clusterName) || oldSelf.clusterName
+                == self.clusterName'
+            - message: clusterName and nominatedClusterNames are mutually exclusive
+              rule: '!has(self.clusterName) || (!has(self.nominatedClusterNames) ||
+                (has(self.nominatedClusterNames) && size(self.nominatedClusterNames)
+                == 0))'
         type: object
         x-kubernetes-validations:
         - message: podSetAssignments must have the same number of podSets as the spec
diff --git a/go.mod b/go.mod
index 1c59e6944..7dc701f4e 100644
--- a/go.mod
+++ b/go.mod
@@ -20,6 +20,7 @@ require (
 	github.com/spf13/cobra v1.9.1
 	github.com/spf13/pflag v1.0.7
 	go.uber.org/zap v1.27.0
+	golang.org/x/exp v0.0.0-20250305212735-054e65f0b394
 	golang.org/x/sync v0.16.0
 	k8s.io/api v0.33.3
 	k8s.io/apimachinery v0.33.3
@@ -122,7 +123,6 @@ require (
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.3 // indirect
 	golang.org/x/crypto v0.38.0 // indirect
-	golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect
 	golang.org/x/mod v0.24.0 // indirect
 	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/oauth2 v0.28.0 // indirect
diff --git a/pkg/cache/snapshot.go b/pkg/cache/snapshot.go
index fb0fdb507..433a400d2 100644
--- a/pkg/cache/snapshot.go
+++ b/pkg/cache/snapshot.go
@@ -181,7 +181,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, penalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) (*ClusterQueueSnapshot, error) {
+func (c *Cache) snapshotClusterQueue(ctx context.Context, cq *clusterQueue, afsEntryPenalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) (*ClusterQueueSnapshot, error) {
 	cc := &ClusterQueueSnapshot{
 		Name:                          cq.Name,
 		ResourceGroups:                make([]ResourceGroup, len(cq.ResourceGroups)),
@@ -210,7 +210,7 @@ func (c *Cache) snapshotClusterQueue(ctx context.Context, cq *clusterQueue, pena
 			return cc, nil
 		}
 		for _, wl := range cc.Workloads {
-			usage, err := wl.CalcLocalQueueFSUsage(ctx, c.client, resourceWeights, penalties)
+			usage, err := wl.CalcLocalQueueFSUsage(ctx, c.client, resourceWeights, afsEntryPenalties)
 			if err != nil {
 				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)})
 			}
diff --git a/pkg/controller/admissionchecks/multikueue/controllers.go b/pkg/controller/admissionchecks/multikueue/controllers.go
index 7eda7795e..9be4c62b6 100644
--- a/pkg/controller/admissionchecks/multikueue/controllers.go
+++ b/pkg/controller/admissionchecks/multikueue/controllers.go
@@ -21,6 +21,7 @@ import (
 
 	ctrl "sigs.k8s.io/controller-runtime"
 
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 )
@@ -37,6 +38,7 @@ type SetupOptions struct {
 	workerLostTimeout time.Duration
 	eventsBatchPeriod time.Duration
 	adapters          map[string]jobframework.MultiKueueAdapter
+	dispatcherName    string
 }
 
 type SetupOption func(o *SetupOptions)
@@ -80,6 +82,13 @@ func WithAdapters(adapters map[string]jobframework.MultiKueueAdapter) SetupOptio
 	}
 }
 
+// WithDispatcherName sets or updates the dispatcher of the MultiKueue workload.
+func WithDispatcherName(dispatcherName string) SetupOption {
+	return func(o *SetupOptions) {
+		o.dispatcherName = dispatcherName
+	}
+}
+
 func SetupControllers(mgr ctrl.Manager, namespace string, opts ...SetupOption) error {
 	options := &SetupOptions{
 		gcInterval:        defaultGCInterval,
@@ -87,6 +96,7 @@ func SetupControllers(mgr ctrl.Manager, namespace string, opts ...SetupOption) e
 		workerLostTimeout: defaultWorkerLostTimeout,
 		eventsBatchPeriod: constants.UpdatesBatchPeriod,
 		adapters:          make(map[string]jobframework.MultiKueueAdapter),
+		dispatcherName:    configapi.MultiKueueDispatcherModeAllAtOnce,
 	}
 
 	for _, o := range opts {
@@ -116,6 +126,7 @@ func SetupControllers(mgr ctrl.Manager, namespace string, opts ...SetupOption) e
 		return err
 	}
 
-	wlRec := newWlReconciler(mgr.GetClient(), helper, cRec, options.origin, mgr.GetEventRecorderFor(constants.WorkloadControllerName), options.workerLostTimeout, options.eventsBatchPeriod, options.adapters)
+	wlRec := newWlReconciler(mgr.GetClient(), helper, cRec, options.origin, mgr.GetEventRecorderFor(constants.WorkloadControllerName),
+		options.workerLostTimeout, options.eventsBatchPeriod, options.adapters, options.dispatcherName)
 	return wlRec.setupWithManager(mgr)
 }
diff --git a/pkg/controller/admissionchecks/multikueue/workload.go b/pkg/controller/admissionchecks/multikueue/workload.go
index 455901617..b0e617dd6 100644
--- a/pkg/controller/admissionchecks/multikueue/workload.go
+++ b/pkg/controller/admissionchecks/multikueue/workload.go
@@ -23,14 +23,18 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-logr/logr"
+	"golang.org/x/exp/slices"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
 	"k8s.io/utils/clock"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -40,6 +44,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
@@ -48,6 +53,10 @@ import (
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
+const (
+	incrementalDispatcherRoundTimeout = 5 * time.Minute
+)
+
 var (
 	realClock           = clock.RealClock{}
 	errNoActiveClusters = errors.New("no active clusters")
@@ -64,6 +73,8 @@ type wlReconciler struct {
 	adapters          map[string]jobframework.MultiKueueAdapter
 	recorder          record.EventRecorder
 	clock             clock.Clock
+	dispatcherName    string
+	roundStartTimes   map[types.NamespacedName]time.Time
 }
 
 var _ reconcile.Reconciler = (*wlReconciler)(nil)
@@ -402,13 +413,16 @@ func (w *wlReconciler) reconcileGroup(ctx context.Context, group *wlGroup) (reco
 			// update the transition time since is used to detect the lost worker state.
 			acs.LastTransitionTime = metav1.NewTime(w.clock.Now())
 
-			wlPatch := workload.BaseSSAWorkload(group.local)
-			workload.SetAdmissionCheckState(&wlPatch.Status.AdmissionChecks, *acs, w.clock)
-			err := w.client.Status().Patch(ctx, wlPatch, client.Apply, client.FieldOwner(kueue.MultiKueueControllerName), client.ForceOwnership)
-			if err != nil {
+			workload.SetAdmissionCheckState(&group.local.Status.AdmissionChecks, *acs, w.clock)
+			// Set the cluster name to the reserving remote and clear the nominated clusters.
+			group.local.Status.ClusterName = &reservingRemote
+			group.local.Status.NominatedClusterNames = nil
+
+			if err := workload.ApplyAdmissionStatus(ctx, w.client, group.local, true, w.clock); err != nil {
+				log.V(2).Error(err, "Failed to patch workload", "workload", klog.KObj(group.local))
 				return reconcile.Result{}, err
 			}
-			w.recorder.Eventf(wlPatch, corev1.EventTypeNormal, "MultiKueue", acs.Message)
+			w.recorder.Eventf(group.local, corev1.EventTypeNormal, "MultiKueue", acs.Message)
 		}
 		return reconcile.Result{RequeueAfter: w.workerLostTimeout}, nil
 	} else if acs.State == kueue.CheckStateReady {
@@ -428,20 +442,102 @@ func (w *wlReconciler) reconcileGroup(ctx context.Context, group *wlGroup) (reco
 		}
 	}
 
-	// finally - create missing workloads
+	return w.nominateAndSynchronizeWorkers(ctx, group)
+}
+
+func (w *wlReconciler) nominateAndSynchronizeWorkers(ctx context.Context, group *wlGroup) (reconcile.Result, error) {
+	log := ctrl.LoggerFrom(ctx).WithValues("op", "nominateAndSynchronizeWorkers")
+	log.V(3).Info("Nominate and Synchronize Worker Clusters")
+	var nominatedWorkers []string
+	var retryAfter time.Duration
+
+	switch w.dispatcherName {
+	case config.MultiKueueDispatcherModeAllAtOnce:
+		for workerName := range group.remotes {
+			nominatedWorkers = append(nominatedWorkers, workerName)
+		}
+		if group.local.Status.ClusterName == nil && !equality.Semantic.DeepEqual(group.local.Status.NominatedClusterNames, nominatedWorkers) {
+			group.local.Status.NominatedClusterNames = nominatedWorkers
+			if err := workload.ApplyAdmissionStatus(ctx, w.client, group.local, true, w.clock); err != nil {
+				log.V(2).Error(err, "Failed to patch nominated clusters", "workload", klog.KObj(group.local))
+				return reconcile.Result{}, err
+			}
+		}
+	case config.MultiKueueDispatcherModeIncremental:
+		key := client.ObjectKeyFromObject(group.local)
+		roundStart, found := w.roundStartTimes[key]
+		now := w.clock.Now()
+
+		if found && now.Sub(roundStart) <= incrementalDispatcherRoundTimeout {
+			remainingWaitTime := incrementalDispatcherRoundTimeout - now.Sub(roundStart)
+			log.V(5).Info("Incremental Dispatcher nomination round still in progress", "remainingWaitTime", remainingWaitTime)
+			return reconcile.Result{RequeueAfter: remainingWaitTime}, nil
+		}
+
+		nextNominatedWorkers, err := getNextNominatedWorkers(log, group)
+		if err != nil {
+			log.Error(err, "nominating next worker clusters failed")
+			return reconcile.Result{}, err
+		}
+
+		nominatedWorkers = append(group.local.Status.NominatedClusterNames, nextNominatedWorkers...)
+		w.roundStartTimes[key] = now
+		group.local.Status.NominatedClusterNames = nominatedWorkers
+		if err := workload.ApplyAdmissionStatus(ctx, w.client, group.local, true, w.clock); err != nil {
+			log.V(2).Error(err, "Failed to patch nominated clusters", "workload", klog.KObj(group.local))
+			return reconcile.Result{}, err
+		}
+	default:
+		log.V(3).Info("External nomination of the worker clusters", "dispatcherName", w.dispatcherName)
+		nominatedWorkers = group.local.Status.NominatedClusterNames
+	}
+
+	log.V(4).Info("Synchronize nominated worker clusters", "nominatedWorkerClusterNames", nominatedWorkers)
+
 	var errs []error
-	for rem, remWl := range group.remotes {
-		if remWl == nil {
-			clone := cloneForCreate(group.local, group.remoteClients[rem].origin)
-			err := group.remoteClients[rem].client.Create(ctx, clone)
-			if err != nil {
-				// just log the error for a single remote
-				log.V(2).Error(err, "creating remote object", "remote", rem)
+	for rem, remoteWl := range group.remotes {
+		if slices.Contains(nominatedWorkers, rem) {
+			if remoteWl == nil {
+				clone := cloneForCreate(group.local, group.remoteClients[rem].origin)
+				if err := group.remoteClients[rem].client.Create(ctx, clone); err != nil {
+					log.V(2).Error(err, "creating remote object", "remote", rem)
+					errs = append(errs, err)
+				}
+			}
+		} else if remoteWl != nil {
+			if err := client.IgnoreNotFound(group.RemoveRemoteObjects(ctx, rem)); err != nil {
+				log.V(2).Error(err, "removing non-nominated remote object", "remote", rem)
 				errs = append(errs, err)
 			}
+			group.remotes[rem] = nil
+		}
+	}
+	return reconcile.Result{RequeueAfter: retryAfter}, errors.Join(errs...)
+}
+
+// getNextNominatedWorkers returns the next set of nominated workers for incremental dispatching.
+// It nominates up to 3 remotes that have not yet been nominated, in sorted order.
+func getNextNominatedWorkers(log logr.Logger, group *wlGroup) ([]string, error) {
+	alreadyNominated := sets.New[string](group.local.Status.NominatedClusterNames...)
+
+	workers := make([]string, 0, len(group.remotes))
+	for remoteWorker := range group.remotes {
+		if !alreadyNominated.Has(remoteWorker) {
+			workers = append(workers, remoteWorker)
 		}
 	}
-	return reconcile.Result{}, errors.Join(errs...)
+	slices.Sort(workers)
+
+	log.V(5).Info("getNextNominatedWorkers (incremental)", "alreadyNominatedClusterNames", alreadyNominated, "remainingClusterNames", workers)
+
+	if len(workers) == 0 {
+		return nil, errors.New("no more workers to nominate")
+	}
+	batchSize := 3
+	if len(workers) < batchSize {
+		return workers, nil
+	}
+	return workers[:batchSize], nil
 }
 
 func (w *wlReconciler) Create(_ event.CreateEvent) bool {
@@ -463,7 +559,11 @@ func (w *wlReconciler) Generic(_ event.GenericEvent) bool {
 	return true
 }
 
-func newWlReconciler(c client.Client, helper *multiKueueStoreHelper, cRec *clustersReconciler, origin string, recorder record.EventRecorder, workerLostTimeout, eventsBatchPeriod time.Duration, adapters map[string]jobframework.MultiKueueAdapter, options ...Option) *wlReconciler {
+func newWlReconciler(c client.Client, helper *multiKueueStoreHelper, cRec *clustersReconciler, origin string,
+	recorder record.EventRecorder, workerLostTimeout, eventsBatchPeriod time.Duration,
+	adapters map[string]jobframework.MultiKueueAdapter, dispatcherName string,
+	options ...Option,
+) *wlReconciler {
 	r := &wlReconciler{
 		client:            c,
 		helper:            helper,
@@ -475,6 +575,8 @@ func newWlReconciler(c client.Client, helper *multiKueueStoreHelper, cRec *clust
 		adapters:          adapters,
 		recorder:          recorder,
 		clock:             realClock,
+		dispatcherName:    dispatcherName,
+		roundStartTimes:   make(map[types.NamespacedName]time.Time),
 	}
 	for _, option := range options {
 		option(r)
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 0a661df3b..639cbc443 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -251,8 +251,11 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			wl.Status.RequeueState = nil
 			updated = true
 		}
-		reportWorkloadEvictedOnce := workload.WorkloadEvictionStateInc(&wl, kueue.WorkloadDeactivated, "")
+
+		// 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, "")
 		if updated {
 			if err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true, r.clock); err != nil {
 				if apierrors.IsNotFound(err) {
diff --git a/pkg/queue/cluster_queue.go b/pkg/queue/cluster_queue.go
index 7d2076fae..bbf70d54d 100644
--- a/pkg/queue/cluster_queue.go
+++ b/pkg/queue/cluster_queue.go
@@ -96,9 +96,9 @@ func workloadKey(i *workload.Info) workload.Reference {
 	return workload.Key(i.Obj)
 }
 
-func newClusterQueue(ctx context.Context, client client.Client, cq *kueue.ClusterQueue, wo workload.Ordering, afsConfig *config.AdmissionFairSharing, penalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) (*ClusterQueue, error) {
+func newClusterQueue(ctx context.Context, client client.Client, cq *kueue.ClusterQueue, wo workload.Ordering, afsConfig *config.AdmissionFairSharing, afsEntryPenalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) (*ClusterQueue, error) {
 	enableAdmissionFs, fsResWeights := afs.ResourceWeights(cq.Spec.AdmissionScope, afsConfig)
-	cqImpl := newClusterQueueImpl(ctx, client, wo, realClock, fsResWeights, enableAdmissionFs, penalties)
+	cqImpl := newClusterQueueImpl(ctx, client, wo, realClock, fsResWeights, enableAdmissionFs, afsEntryPenalties)
 	err := cqImpl.Update(cq)
 	if err != nil {
 		return nil, err
@@ -106,8 +106,8 @@ func newClusterQueue(ctx context.Context, client client.Client, cq *kueue.Cluste
 	return cqImpl, nil
 }
 
-func newClusterQueueImpl(ctx context.Context, client client.Client, wo workload.Ordering, clock clock.Clock, fsResWeights map[corev1.ResourceName]float64, enableAdmissionFs bool, penalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) *ClusterQueue {
-	lessFunc := queueOrderingFunc(ctx, client, wo, fsResWeights, enableAdmissionFs, penalties)
+func newClusterQueueImpl(ctx context.Context, client client.Client, wo workload.Ordering, clock clock.Clock, fsResWeights map[corev1.ResourceName]float64, enableAdmissionFs bool, afsEntryPenalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) *ClusterQueue {
+	lessFunc := queueOrderingFunc(ctx, client, wo, fsResWeights, enableAdmissionFs, afsEntryPenalties)
 	return &ClusterQueue{
 		heap:                   *heap.New(workloadKey, lessFunc),
 		inadmissibleWorkloads:  make(map[workload.Reference]*workload.Info),
@@ -428,12 +428,12 @@ func (c *ClusterQueue) RequeueIfNotPresent(wInfo *workload.Info, reason RequeueR
 // to sort workloads. The function sorts workloads based on their priority.
 // When priorities are equal, it uses the workload's creation or eviction
 // time.
-func queueOrderingFunc(ctx context.Context, c client.Client, wo workload.Ordering, fsResWeights map[corev1.ResourceName]float64, enableAdmissionFs bool, penalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) func(a, b *workload.Info) bool {
+func queueOrderingFunc(ctx context.Context, c client.Client, wo workload.Ordering, fsResWeights map[corev1.ResourceName]float64, enableAdmissionFs bool, afsEntryPenalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) func(a, b *workload.Info) bool {
 	log := ctrl.LoggerFrom(ctx)
 	return func(a, b *workload.Info) bool {
 		if enableAdmissionFs {
-			lqAUsage, errA := a.CalcLocalQueueFSUsage(ctx, c, fsResWeights, penalties)
-			lqBUsage, errB := b.CalcLocalQueueFSUsage(ctx, c, fsResWeights, penalties)
+			lqAUsage, errA := a.CalcLocalQueueFSUsage(ctx, c, fsResWeights, afsEntryPenalties)
+			lqBUsage, errB := b.CalcLocalQueueFSUsage(ctx, c, fsResWeights, afsEntryPenalties)
 			switch {
 			case errA != nil:
 				log.V(2).Error(errA, "Error determining LocalQueue usage")
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index 437f7e1a1..a5fb216b6 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -181,11 +181,11 @@ func (m *Manager) AddClusterQueue(ctx context.Context, cq *kueue.ClusterQueue) e
 		return errClusterQueueAlreadyExists
 	}
 
-	var penalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]
+	var afsEntryPenalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]
 	if features.Enabled(features.AdmissionFairSharing) {
-		penalties = m.afsEntryPenalties.GetPenalties()
+		afsEntryPenalties = m.afsEntryPenalties.GetPenalties()
 	}
-	cqImpl, err := newClusterQueue(ctx, m.client, cq, m.workloadOrdering, m.admissionFairSharingConfig, penalties)
+	cqImpl, err := newClusterQueue(ctx, m.client, cq, m.workloadOrdering, m.admissionFairSharingConfig, afsEntryPenalties)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 4ad72da2d..4c03b612c 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -194,7 +194,7 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 	log := ctrl.LoggerFrom(ctx).WithValues("schedulingCycle", s.schedulingCycle)
 	ctx = ctrl.LoggerInto(ctx, log)
 
-	snapshotOpts := make([]cache.SnapshotOption, 0)
+	var snapshotOpts []cache.SnapshotOption
 	if features.Enabled(features.AdmissionFairSharing) && s.queues.HasAnyEntryPenalty() {
 		s.queues.HeapifyAllClusterQueues()
 		snapshotOpts = append(snapshotOpts, cache.WithAfsEntryPenalties(s.queues.GetAfsEntryPenalties()))
@@ -623,7 +623,7 @@ func (s *Scheduler) admit(ctx context.Context, e *entry, cq *cache.ClusterQueueS
 	log.V(2).Info("Workload assumed in the cache")
 
 	if features.Enabled(features.AdmissionFairSharing) {
-		s.addEntryPenaltyToLq(log, e)
+		s.updateEntryPenalty(log, e, add)
 
 		// Trigger LocalQueue reconciler to apply any pending penalties
 		s.queues.NotifyWorkloadUpdateWatchers(e.Obj, newWorkload)
@@ -642,7 +642,7 @@ func (s *Scheduler) admit(ctx context.Context, e *entry, cq *cache.ClusterQueueS
 		// by an event.
 		_ = s.cache.ForgetWorkload(log, newWorkload)
 		if features.Enabled(features.AdmissionFairSharing) {
-			s.subEntryPenaltyToLq(log, e)
+			s.updateEntryPenalty(log, e, subtract)
 		}
 		if apierrors.IsNotFound(err) {
 			log.V(2).Info("Workload not admitted because it was deleted")
@@ -852,18 +852,25 @@ func (s *Scheduler) replaceWorkloadSlice(ctx context.Context, oldQueue kueue.Clu
 	return nil
 }
 
-func (s *Scheduler) addEntryPenaltyToLq(log logr.Logger, e *entry) {
-	lqKey := utilqueue.NewLocalQueueReference(e.Obj.Namespace, e.Obj.Spec.QueueName)
-	penalty := afs.CalculateEntryPenalty(e.SumTotalRequests(), s.admissionFairSharing)
-	s.queues.PushEntryPenalty(lqKey, penalty)
+type usageOp int
 
-	log.V(3).Info("Entry penalty added to lq", "lqKey", lqKey, "penalty", penalty)
-}
+const (
+	// add penalty
+	add usageOp = iota
+	// subtract penalty
+	subtract
+)
 
-func (s *Scheduler) subEntryPenaltyToLq(log logr.Logger, e *entry) {
+func (s *Scheduler) updateEntryPenalty(log logr.Logger, e *entry, op usageOp) {
 	lqKey := utilqueue.NewLocalQueueReference(e.Obj.Namespace, e.Obj.Spec.QueueName)
 	penalty := afs.CalculateEntryPenalty(e.SumTotalRequests(), s.admissionFairSharing)
-	s.queues.SubEntryPenalty(lqKey, penalty)
 
-	log.V(3).Info("Entry penalty subtracted from lq", "lqKey", lqKey, "penalty", penalty)
+	switch op {
+	case add:
+		s.queues.PushEntryPenalty(lqKey, penalty)
+		log.V(3).Info("Entry penalty added to lq", "lqKey", lqKey, "penalty", penalty)
+	case subtract:
+		s.queues.SubEntryPenalty(lqKey, penalty)
+		log.V(3).Info("Entry penalty subtracted from lq", "lqKey", lqKey, "penalty", penalty)
+	}
 }
diff --git a/pkg/webhooks/webhooks.go b/pkg/webhooks/webhooks.go
index cb7c832bb..c367d94d0 100644
--- a/pkg/webhooks/webhooks.go
+++ b/pkg/webhooks/webhooks.go
@@ -22,8 +22,8 @@ import (
 
 // Setup sets up the webhooks for core controllers. It returns the name of the
 // webhook that failed to create and an error, if any.
-func Setup(mgr ctrl.Manager) (string, error) {
-	if err := setupWebhookForWorkload(mgr); err != nil {
+func Setup(mgr ctrl.Manager, dispatcherName string) (string, error) {
+	if err := setupWebhookForWorkload(mgr, dispatcherName); err != nil {
 		return "Workload", err
 	}
 
diff --git a/pkg/webhooks/workload_webhook.go b/pkg/webhooks/workload_webhook.go
index bb2e8a50e..e72170077 100644
--- a/pkg/webhooks/workload_webhook.go
+++ b/pkg/webhooks/workload_webhook.go
@@ -19,6 +19,7 @@ package webhooks
 import (
 	"context"
 	"fmt"
+	"slices"
 	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
@@ -32,20 +33,26 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/resources"
-	"sigs.k8s.io/kueue/pkg/util/slices"
+	utilslices "sigs.k8s.io/kueue/pkg/util/slices"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
-type WorkloadWebhook struct{}
+type WorkloadWebhook struct {
+	dispatcherName string
+}
 
-func setupWebhookForWorkload(mgr ctrl.Manager) error {
+func setupWebhookForWorkload(mgr ctrl.Manager, dispatcherName string) error {
+	wh := &WorkloadWebhook{
+		dispatcherName: dispatcherName,
+	}
 	return ctrl.NewWebhookManagedBy(mgr).
 		For(&kueue.Workload{}).
-		WithDefaulter(&WorkloadWebhook{}).
-		WithValidator(&WorkloadWebhook{}).
+		WithDefaulter(wh).
+		WithValidator(wh).
 		Complete()
 }
 
@@ -87,7 +94,7 @@ func (w *WorkloadWebhook) ValidateUpdate(ctx context.Context, oldObj, newObj run
 	oldWL := oldObj.(*kueue.Workload)
 	log := ctrl.LoggerFrom(ctx).WithName("workload-webhook")
 	log.V(5).Info("Validating update")
-	return nil, ValidateWorkloadUpdate(newWL, oldWL).ToAggregate()
+	return nil, ValidateWorkloadUpdate(newWL, oldWL, w.dispatcherName).ToAggregate()
 }
 
 // ValidateDelete implements webhook.CustomValidator so a webhook will be registered for the type
@@ -165,7 +172,7 @@ func validateAdmissionChecks(obj *kueue.Workload, basePath *field.Path) field.Er
 func validatePodSetUpdates(acs *kueue.AdmissionCheckState, obj *kueue.Workload, basePath *field.Path) field.ErrorList {
 	var allErrs field.ErrorList
 
-	knowPodSets := sets.New(slices.Map(obj.Spec.PodSets, func(ps *kueue.PodSet) kueue.PodSetReference {
+	knowPodSets := sets.New(utilslices.Map(obj.Spec.PodSets, func(ps *kueue.PodSet) kueue.PodSetReference {
 		return ps.Name
 	})...)
 
@@ -185,7 +192,7 @@ func validatePodSetUpdates(acs *kueue.AdmissionCheckState, obj *kueue.Workload,
 
 func validateImmutablePodSetUpdates(newObj, oldObj *kueue.Workload, basePath *field.Path) field.ErrorList {
 	var allErrs field.ErrorList
-	newAcs := slices.ToRefMap(newObj.Status.AdmissionChecks, func(f *kueue.AdmissionCheckState) kueue.AdmissionCheckReference { return f.Name })
+	newAcs := utilslices.ToRefMap(newObj.Status.AdmissionChecks, func(f *kueue.AdmissionCheckState) kueue.AdmissionCheckReference { return f.Name })
 	for i := range oldObj.Status.AdmissionChecks {
 		oldAc := &oldObj.Status.AdmissionChecks[i]
 		newAc, found := newAcs[oldAc.Name]
@@ -266,7 +273,7 @@ func validateReclaimablePods(obj *kueue.Workload, basePath *field.Path) field.Er
 	return ret
 }
 
-func ValidateWorkloadUpdate(newObj, oldObj *kueue.Workload) field.ErrorList {
+func ValidateWorkloadUpdate(newObj, oldObj *kueue.Workload, dispatcherName string) field.ErrorList {
 	var allErrs field.ErrorList
 	specPath := field.NewPath("spec")
 	statusPath := field.NewPath("status")
@@ -280,7 +287,7 @@ func ValidateWorkloadUpdate(newObj, oldObj *kueue.Workload) field.ErrorList {
 	}
 	allErrs = append(allErrs, validateAdmissionUpdate(newObj.Status.Admission, oldObj.Status.Admission, field.NewPath("status", "admission"))...)
 	allErrs = append(allErrs, validateImmutablePodSetUpdates(newObj, oldObj, statusPath.Child("admissionChecks"))...)
-
+	allErrs = append(allErrs, validateClusterNameUpdate(newObj, oldObj, dispatcherName, statusPath)...)
 	return allErrs
 }
 
@@ -366,3 +373,15 @@ func validateImmutablePodSets(new, old []kueue.PodSet, path *field.Path) field.E
 	}
 	return allErrs
 }
+
+func validateClusterNameUpdate(newObj, oldObj *kueue.Workload, dispatcherName string, statusPath *field.Path) field.ErrorList {
+	var allErrs field.ErrorList
+	if oldObj.Status.ClusterName == nil && newObj.Status.ClusterName != nil && dispatcherName != configapi.MultiKueueDispatcherModeAllAtOnce {
+		found := slices.Contains(oldObj.Status.NominatedClusterNames, *newObj.Status.ClusterName)
+		if !found {
+			allErrs = append(allErrs, field.Invalid(statusPath.Child("clusterName"), newObj.Status.ClusterName, "when setting clusterName it must be one of the nominatedClusterNames"))
+		}
+	}
+
+	return allErrs
+}
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 07f8ca411..a875507d4 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -302,7 +302,7 @@ func dropExcludedResources(input corev1.ResourceList, excludedPrefixes []string)
 	return res
 }
 
-func (i *Info) CalcLocalQueueFSUsage(ctx context.Context, c client.Client, resWeights map[corev1.ResourceName]float64, penalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) (float64, error) {
+func (i *Info) CalcLocalQueueFSUsage(ctx context.Context, c client.Client, resWeights map[corev1.ResourceName]float64, afsEntryPenalties *utilmaps.SyncMap[utilqueue.LocalQueueReference, corev1.ResourceList]) (float64, error) {
 	var lq kueue.LocalQueue
 	lqKey := client.ObjectKey{Namespace: i.Obj.Namespace, Name: string(i.Obj.Spec.QueueName)}
 	if err := c.Get(ctx, lqKey, &lq); err != nil {
@@ -321,8 +321,8 @@ func (i *Info) CalcLocalQueueFSUsage(ctx context.Context, c client.Client, resWe
 		usage += weight * resVal.AsApproximateFloat64()
 	}
 	penalty := corev1.ResourceList{}
-	if penalties != nil {
-		penalty, _ = penalties.Get(utilqueue.Key(&lq))
+	if afsEntryPenalties != nil {
+		penalty, _ = afsEntryPenalties.Get(utilqueue.Key(&lq))
 	}
 	for resName, penaltyVal := range penalty {
 		weight, found := resWeights[resName]
@@ -823,6 +823,8 @@ func AdmissionStatusPatch(w *kueue.Workload, wlCopy *kueue.Workload, strict bool
 		}
 		wlCopy.Status.SchedulingStats.Evictions = append(wlCopy.Status.SchedulingStats.Evictions, w.Status.SchedulingStats.Evictions...)
 	}
+	wlCopy.Status.ClusterName = w.Status.ClusterName
+	wlCopy.Status.NominatedClusterNames = w.Status.NominatedClusterNames
 }
 
 func AdmissionChecksStatusPatch(w *kueue.Workload, wlCopy *kueue.Workload, c clock.Clock) {
@@ -1093,9 +1095,15 @@ func EvictWorkload(ctx context.Context, c client.Client, recorder record.EventRe
 
 func PrepareForEviction(w *kueue.Workload, now time.Time, reason, message string) bool {
 	SetEvictedCondition(w, reason, message)
+	ResetClusterNomination(w)
 	return ResetChecksOnEviction(w, now)
 }
 
+func ResetClusterNomination(w *kueue.Workload) {
+	w.Status.ClusterName = nil
+	w.Status.NominatedClusterNames = nil
+}
+
 func ReportEvictedWorkload(recorder record.EventRecorder, wl *kueue.Workload, cqName kueue.ClusterQueueReference, reason, message string) {
 	durationToPreemption := workloadsWithPodsReadyToEvictedTime(wl)
 	metrics.ReportEvictedWorkloads(cqName, reason, durationToPreemption)
diff --git a/site/content/en/docs/reference/kueue-config.v1beta1.md b/site/content/en/docs/reference/kueue-config.v1beta1.md
index 2e9bb4147..c0aa05b44 100644
--- a/site/content/en/docs/reference/kueue-config.v1beta1.md
+++ b/site/content/en/docs/reference/kueue-config.v1beta1.md
@@ -708,6 +708,17 @@ if the connection with its reserving worker cluster is lost.</p>
 <p>Defaults to 15 minutes.</p>
 </td>
 </tr>
+<tr><td><code>dispatcherName</code><br/>
+<code>string</code>
+</td>
+<td>
+   <p>DispatcherName defines the dispatcher responsible for selecting worker clusters to handle the workload.</p>
+<ul>
+<li>If specified, the workload will be handled by the named dispatcher.</li>
+<li>If not specified, the workload will be handled by the default (&quot;kueue.x-k8s.io/multikueue-dispatcher-all-at-once&quot;) dispatcher.</li>
+</ul>
+</td>
+</tr>
 </tbody>
 </table>
 
diff --git a/site/content/en/docs/reference/kueue.v1beta1.md b/site/content/en/docs/reference/kueue.v1beta1.md
index 9f750d2d4..b511c68b1 100644
--- a/site/content/en/docs/reference/kueue.v1beta1.md
+++ b/site/content/en/docs/reference/kueue.v1beta1.md
@@ -3167,6 +3167,24 @@ in Admitted state, in the previous <code>Admit</code> - <code>Evict</code> cycle
    <p>schedulingStats tracks scheduling statistics</p>
 </td>
 </tr>
+<tr><td><code>nominatedClusterNames</code><br/>
+<code>[]string</code>
+</td>
+<td>
+   <p>nominatedClusterNames specifies the list of cluster names that have been nominated for scheduling.
+This field is mutually exclusive with the <code>.status.clusterName</code> field, and is reset when
+<code>status.clusterName</code> is set.
+This field is optional.</p>
+</td>
+</tr>
+<tr><td><code>clusterName</code><br/>
+<code>string</code>
+</td>
+<td>
+   <p>clusterName is the name of the cluster where the workload is actually assigned.
+This field is reset after the Workload is evicted.</p>
+</td>
+</tr>
 </tbody>
 </table>
   
\ No newline at end of file


Test patch:
diff --git a/apis/config/v1beta1/defaults_test.go b/apis/config/v1beta1/defaults_test.go
index 857c85cbf..382a61ab0 100644
--- a/apis/config/v1beta1/defaults_test.go
+++ b/apis/config/v1beta1/defaults_test.go
@@ -98,6 +98,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 		GCInterval:        &metav1.Duration{Duration: DefaultMultiKueueGCInterval},
 		Origin:            ptr.To(DefaultMultiKueueOrigin),
 		WorkerLostTimeout: &metav1.Duration{Duration: DefaultMultiKueueWorkerLostTimeout},
+		DispatcherName:    ptr.To(MultiKueueDispatcherModeAllAtOnce),
 	}
 
 	podsReadyTimeout := metav1.Duration{Duration: defaultPodsReadyTimeout}
@@ -530,6 +531,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					GCInterval:        &metav1.Duration{Duration: time.Second},
 					Origin:            ptr.To("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
+					DispatcherName:    ptr.To[string](MultiKueueDispatcherModeIncremental),
 				},
 			},
 			want: &Configuration{
@@ -545,6 +547,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					GCInterval:        &metav1.Duration{Duration: time.Second},
 					Origin:            ptr.To("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
+					DispatcherName:    ptr.To[string](MultiKueueDispatcherModeIncremental),
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				WaitForPodsReady:             &WaitForPodsReady{},
@@ -559,6 +562,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					GCInterval:        &metav1.Duration{Duration: time.Second},
 					Origin:            ptr.To(""),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
+					DispatcherName:    defaultMultiKueue.DispatcherName,
 				},
 			},
 			want: &Configuration{
@@ -574,6 +578,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					GCInterval:        &metav1.Duration{Duration: time.Second},
 					Origin:            ptr.To(DefaultMultiKueueOrigin),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
+					DispatcherName:    defaultMultiKueue.DispatcherName,
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				WaitForPodsReady:             &WaitForPodsReady{},
@@ -602,6 +607,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					GCInterval:        &metav1.Duration{},
 					Origin:            ptr.To("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: 15 * time.Minute},
+					DispatcherName:    defaultMultiKueue.DispatcherName,
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				WaitForPodsReady:             &WaitForPodsReady{},
diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go
index bef95dad2..300fe9a63 100644
--- a/pkg/config/config_test.go
+++ b/pkg/config/config_test.go
@@ -281,6 +281,7 @@ multiKueue:
   gcInterval: 1m30s
   origin: multikueue-manager1
   workerLostTimeout: 10m
+  dispatcherName: kueue.x-k8s.io/multikueue-dispatcher-incremental
 `), os.FileMode(0600)); err != nil {
 		t.Fatal(err)
 	}
@@ -411,6 +412,7 @@ objectRetentionPolicies:
 		GCInterval:        &metav1.Duration{Duration: configapi.DefaultMultiKueueGCInterval},
 		Origin:            ptr.To(configapi.DefaultMultiKueueOrigin),
 		WorkerLostTimeout: &metav1.Duration{Duration: configapi.DefaultMultiKueueWorkerLostTimeout},
+		DispatcherName:    ptr.To[string](configapi.MultiKueueDispatcherModeAllAtOnce),
 	}
 
 	defaultWaitForPodsReady := &configapi.WaitForPodsReady{}
@@ -906,6 +908,7 @@ objectRetentionPolicies:
 					GCInterval:        &metav1.Duration{Duration: 90 * time.Second},
 					Origin:            ptr.To("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: 10 * time.Minute},
+					DispatcherName:    ptr.To[string](configapi.MultiKueueDispatcherModeIncremental),
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				WaitForPodsReady:             defaultWaitForPodsReady,
@@ -1102,6 +1105,7 @@ func TestEncode(t *testing.T) {
 					"gcInterval":        "1m0s",
 					"origin":            "multikueue",
 					"workerLostTimeout": "15m0s",
+					"dispatcherName":    string(configapi.MultiKueueDispatcherModeAllAtOnce),
 				},
 				"waitForPodsReady": map[string]any{},
 			},
diff --git a/pkg/controller/admissionchecks/multikueue/workload_test.go b/pkg/controller/admissionchecks/multikueue/workload_test.go
index 36a5971d6..031aebd1f 100644
--- a/pkg/controller/admissionchecks/multikueue/workload_test.go
+++ b/pkg/controller/admissionchecks/multikueue/workload_test.go
@@ -19,6 +19,8 @@ package multikueue
 import (
 	"context"
 	"errors"
+	"fmt"
+	"sort"
 	"testing"
 	"time"
 
@@ -30,11 +32,14 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	testingclock "k8s.io/utils/clock/testing"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
@@ -75,6 +80,7 @@ func TestWlReconcile(t *testing.T) {
 		worker1Workloads         []kueue.Workload
 		worker1Jobs              []batchv1.Job
 		withoutJobManagedBy      bool
+		dispatcherName           *string
 
 		// second worker
 		useSecondWorker      bool
@@ -331,6 +337,7 @@ func TestWlReconcile(t *testing.T) {
 					AdmissionCheck(kueue.AdmissionCheckState{Name: "ac1", State: kueue.CheckStatePending}).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
 					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					NominatedClusterNames([]string{"worker1", "worker2"}).
 					Obj(),
 			},
 			wantWorker1Workloads: []kueue.Workload{
@@ -363,6 +370,7 @@ func TestWlReconcile(t *testing.T) {
 					AdmissionCheck(kueue.AdmissionCheckState{Name: "ac1", State: kueue.CheckStatePending}).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
 					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					NominatedClusterNames([]string{"worker1", "worker2"}).
 					Obj(),
 			},
 			wantWorker1Workloads: []kueue.Workload{
@@ -460,6 +468,7 @@ func TestWlReconcile(t *testing.T) {
 					}).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
 					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					ClusterName("worker1").
 					Obj(),
 			},
 			wantManagersJobs: []batchv1.Job{
@@ -524,6 +533,73 @@ func TestWlReconcile(t *testing.T) {
 					}).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
 					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					ClusterName("worker1").
+					Obj(),
+			},
+			wantManagersJobs: []batchv1.Job{
+				*baseJobBuilder.Clone().Obj(),
+			},
+
+			wantWorker1Workloads: []kueue.Workload{
+				*baseWorkloadBuilder.Clone().
+					Label(kueue.MultiKueueOriginLabel, defaultOrigin).
+					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					Obj(),
+			},
+			wantWorker1Jobs: []batchv1.Job{
+				*baseJobBuilder.Clone().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, defaultOrigin).
+					Obj(),
+			},
+
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       client.ObjectKeyFromObject(baseWorkloadBuilder.Clone().Obj()),
+					EventType: "Normal",
+					Reason:    "MultiKueue",
+					Message:   `The workload got reservation on "worker1"`,
+				},
+			},
+		},
+		"remote wl with reservation (withoutJobManagedBy, MultiKueueDispatcherModeIncremental)": {
+			reconcileFor:        "wl1",
+			withoutJobManagedBy: true,
+			dispatcherName:      ptr.To(config.MultiKueueDispatcherModeIncremental),
+			managersWorkloads: []kueue.Workload{
+				*baseWorkloadBuilder.Clone().
+					AdmissionCheck(kueue.AdmissionCheckState{Name: "ac1", State: kueue.CheckStatePending}).
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
+					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					Obj(),
+			},
+
+			managersJobs: []batchv1.Job{
+				*baseJobBuilder.Clone().Obj(),
+			},
+
+			worker1Workloads: []kueue.Workload{
+				*baseWorkloadBuilder.Clone().
+					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					Label(kueue.MultiKueueOriginLabel, defaultOrigin).
+					Obj(),
+			},
+			useSecondWorker: true,
+			worker2Workloads: []kueue.Workload{
+				*baseWorkloadBuilder.Clone().
+					Obj(),
+			},
+
+			wantManagersWorkloads: []kueue.Workload{
+				*baseWorkloadBuilder.Clone().
+					AdmissionCheck(kueue.AdmissionCheckState{
+						Name:    "ac1",
+						State:   kueue.CheckStatePending,
+						Message: `The workload got reservation on "worker1"`,
+					}).
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
+					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					ClusterName("worker1").
 					Obj(),
 			},
 			wantManagersJobs: []batchv1.Job{
@@ -592,6 +668,7 @@ func TestWlReconcile(t *testing.T) {
 					}).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
 					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					ClusterName("worker1").
 					Obj(),
 			},
 			wantManagersJobs: []batchv1.Job{
@@ -663,6 +740,7 @@ func TestWlReconcile(t *testing.T) {
 					}).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
 					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					ClusterName("worker1").
 					Obj(),
 			},
 			wantManagersJobs: []batchv1.Job{
@@ -1018,6 +1096,7 @@ func TestWlReconcile(t *testing.T) {
 					}).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "job1", "uid1").
 					ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+					ClusterName("worker1").
 					Obj(),
 			},
 			wantManagersJobs: []batchv1.Job{
@@ -1115,7 +1194,8 @@ func TestWlReconcile(t *testing.T) {
 
 			helper, _ := newMultiKueueStoreHelper(managerClient)
 			recorder := &utiltesting.EventRecorder{}
-			reconciler := newWlReconciler(managerClient, helper, cRec, defaultOrigin, recorder, defaultWorkerLostTimeout, time.Second, adapters, WithClock(t, fakeClock))
+			mkDispatcherName := ptr.Deref(tc.dispatcherName, config.MultiKueueDispatcherModeAllAtOnce)
+			reconciler := newWlReconciler(managerClient, helper, cRec, defaultOrigin, recorder, defaultWorkerLostTimeout, time.Second, adapters, mkDispatcherName, WithClock(t, fakeClock))
 
 			for _, val := range tc.managersDeletedWorkloads {
 				reconciler.Delete(event.DeleteEvent{
@@ -1136,6 +1216,13 @@ func TestWlReconcile(t *testing.T) {
 			if err := managerClient.List(t.Context(), gotManagersWorkloads); err != nil {
 				t.Errorf("unexpected list manager's workloads error: %s", err)
 			} else {
+				// ensure deterministic comparison
+				for i := range gotManagersWorkloads.Items {
+					sort.Strings(gotManagersWorkloads.Items[i].Status.NominatedClusterNames)
+				}
+				for i := range tc.wantManagersWorkloads {
+					sort.Strings(tc.wantManagersWorkloads[i].Status.NominatedClusterNames)
+				}
 				if diff := cmp.Diff(tc.wantManagersWorkloads, gotManagersWorkloads.Items, objCheckOpts...); diff != "" {
 					t.Errorf("unexpected manager's workloads (-want/+got):\n%s", diff)
 				}
@@ -1194,3 +1281,209 @@ func TestWlReconcile(t *testing.T) {
 		})
 	}
 }
+
+type createCall struct {
+	cluster string
+	obj     *kueue.Workload
+}
+
+func TestNominateAndSynchronizeWorkers_MoreCases(t *testing.T) {
+	const externalMultiKueueDispatcherController = "external.com/mk-dispatcher"
+
+	remoteNames := make([]string, 9)
+	for i := range 9 {
+		remoteNames[i] = fmt.Sprintf("remote%d", i+1)
+	}
+	remotes := make(map[string]*kueue.Workload, len(remoteNames))
+	for _, name := range remoteNames {
+		remotes[name] = nil // initially no workloads on remotes
+	}
+	now := time.Now()
+
+	tests := []struct {
+		name             string
+		dispatcherMode   string
+		remotes          map[string]*kueue.Workload
+		nominatedWorkers []string
+		cond             *metav1.Condition
+		createErr        error
+		wantCreated      []string
+		wantErr          bool
+		advanceClock     time.Duration
+		wantRetryAfter   time.Duration
+	}{
+		{
+			name:           "AllClusters: clone to all remotes, nominates all",
+			dispatcherMode: config.MultiKueueDispatcherModeAllAtOnce,
+			remotes:        map[string]*kueue.Workload{remoteNames[0]: nil, remoteNames[1]: nil},
+			wantCreated:    []string{remoteNames[0], remoteNames[1]},
+			wantRetryAfter: 0,
+		},
+		{
+			name:           "AllClusters: workloads already created on remotes, do not create again",
+			dispatcherMode: config.MultiKueueDispatcherModeAllAtOnce,
+			remotes:        map[string]*kueue.Workload{remoteNames[0]: {}, remoteNames[1]: {}},
+			wantCreated:    nil,
+			wantRetryAfter: 0,
+		},
+		{
+			name:           "Incremental: only one remote, nominates it",
+			dispatcherMode: config.MultiKueueDispatcherModeIncremental,
+			remotes:        map[string]*kueue.Workload{remoteNames[0]: nil},
+			wantCreated:    []string{remoteNames[0]},
+			wantRetryAfter: 0,
+		},
+		{
+			name:             "Incremental: no previous nomination, nominates remote1...3",
+			dispatcherMode:   config.MultiKueueDispatcherModeIncremental,
+			remotes:          remotes,
+			nominatedWorkers: []string{""},
+			wantCreated:      remoteNames[:3],
+			wantRetryAfter:   0,
+		},
+		{
+			name:             "Incremental: keep remote1..3, nomination round still in progress",
+			dispatcherMode:   config.MultiKueueDispatcherModeIncremental,
+			remotes:          remotes,
+			nominatedWorkers: remoteNames[:3],
+			wantCreated:      nil,
+			advanceClock:     incrementalDispatcherRoundTimeout / 2,
+			wantRetryAfter:   incrementalDispatcherRoundTimeout / 2,
+		},
+		{
+			name:             "Incremental: previously remote1..3, nomination round exceeded - increment",
+			dispatcherMode:   config.MultiKueueDispatcherModeIncremental,
+			remotes:          remotes,
+			nominatedWorkers: remoteNames[:3],
+			advanceClock:     incrementalDispatcherRoundTimeout + time.Second,
+			wantCreated:      remoteNames[:6],
+			wantRetryAfter:   0,
+		},
+		{
+			name:             "Incremental: previously remote1..6, next nomination round all clusters",
+			dispatcherMode:   config.MultiKueueDispatcherModeIncremental,
+			remotes:          remotes,
+			nominatedWorkers: remoteNames[:6],
+			advanceClock:     incrementalDispatcherRoundTimeout + time.Second,
+			wantCreated:      remoteNames,
+			wantRetryAfter:   0,
+		},
+		{
+			name:           "External controller: no nominated workers, nothing created",
+			dispatcherMode: externalMultiKueueDispatcherController,
+			remotes:        remotes,
+			wantCreated:    nil,
+			wantRetryAfter: 0,
+		},
+		{
+			name:             "External controller: nominate remote1 and remote6",
+			dispatcherMode:   externalMultiKueueDispatcherController,
+			remotes:          remotes,
+			nominatedWorkers: []string{remoteNames[0], remoteNames[5]},
+			wantCreated:      []string{remoteNames[0], remoteNames[5]},
+		},
+		{
+			name:             "External controller: nominate all remotes at once",
+			dispatcherMode:   externalMultiKueueDispatcherController,
+			remotes:          remotes,
+			nominatedWorkers: remoteNames,
+			wantCreated:      remoteNames,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fakeClock := testingclock.NewFakeClock(now)
+
+			local := &kueue.Workload{
+				ObjectMeta: metav1.ObjectMeta{Name: "wl", Namespace: "ns"},
+				Status: kueue.WorkloadStatus{
+					Conditions:            make([]metav1.Condition, 0, 1),
+					NominatedClusterNames: tt.nominatedWorkers,
+				},
+			}
+
+			var created []createCall
+			makeFakeCreate := func(origin string) func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error {
+				return func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error {
+					created = append(created, createCall{
+						cluster: origin,
+						obj:     obj.(*kueue.Workload),
+					})
+					return c.Create(ctx, obj, opts...)
+				}
+			}
+			objs := []client.Object{local}
+			wlClientBuilder := utiltesting.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+				SubResourcePatch: func(ctx context.Context, client client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+					local.Status.NominatedClusterNames = obj.(*kueue.Workload).Status.NominatedClusterNames
+					return utiltesting.TreatSSAAsStrategicMerge(ctx, client, subResourceName, obj, patch, opts...)
+				},
+			}).WithObjects(objs...).WithStatusSubresource(objs...)
+
+			remoteClientBuilders := make(map[string]*fake.ClientBuilder, len(tt.remotes))
+			for remote := range tt.remotes {
+				remoteClientBuilders[remote] = utiltesting.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+					Create: makeFakeCreate(remote),
+				},
+				)
+			}
+
+			remoteClients := make(map[string]*remoteClient, len(tt.remotes))
+			for remote, builder := range remoteClientBuilders {
+				remoteClients[remote] = &remoteClient{client: builder.Build(), origin: remote}
+			}
+
+			if tt.cond != nil {
+				local.Status.Conditions = append(local.Status.Conditions, *tt.cond)
+			}
+			group := &wlGroup{
+				local:         local,
+				remotes:       tt.remotes,
+				remoteClients: remoteClients,
+				acName:        "ac1",
+			}
+
+			wlRec := &wlReconciler{
+				clock:           fakeClock,
+				dispatcherName:  tt.dispatcherMode,
+				client:          wlClientBuilder.Build(),
+				roundStartTimes: make(map[types.NamespacedName]time.Time),
+			}
+
+			if tt.advanceClock > 0 {
+				key := types.NamespacedName{Name: group.local.Name, Namespace: group.local.Namespace}
+				wlRec.roundStartTimes[key] = fakeClock.Now()
+				fakeClock.SetTime(now.Add(tt.advanceClock))
+			}
+
+			res, err := wlRec.nominateAndSynchronizeWorkers(t.Context(), group)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("expected error: %v, got: %v", tt.wantErr, err)
+			}
+
+			var gotCreated []string
+			for _, c := range created {
+				gotCreated = append(gotCreated, c.cluster)
+			}
+			s1 := sort.StringSlice(tt.wantCreated)
+			s1.Sort()
+			s2 := sort.StringSlice(gotCreated)
+			s2.Sort()
+			if diff := cmp.Diff(s1, s2); diff != "" {
+				t.Errorf("unexpected created remotes (-want/+got):\n%s", diff)
+			}
+
+			const timeMargin = 100 * time.Millisecond
+			if tt.wantRetryAfter == 0 {
+				if res.RequeueAfter != 0 {
+					t.Errorf("unexpected RequeueAfter, want %v, got %v", tt.wantRetryAfter, res.RequeueAfter)
+				}
+			} else {
+				if res.RequeueAfter < tt.wantRetryAfter-timeMargin || res.RequeueAfter > tt.wantRetryAfter+timeMargin {
+					t.Errorf("unexpected RequeueAfter, want %v±%v, got %v", tt.wantRetryAfter, timeMargin, res.RequeueAfter)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller_test.go b/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller_test.go
index 25cb7f73a..f8479bd57 100644
--- a/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller_test.go
+++ b/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller_test.go
@@ -30,6 +30,7 @@ import (
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
@@ -569,7 +570,7 @@ func TestReconciler(t *testing.T) {
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
 			ctx, _ := utiltesting.ContextWithLog(t)
-			kcBuilder := utiltesting.NewClientBuilder(kftraining.AddToScheme)
+			kcBuilder := utiltesting.NewClientBuilder(kftraining.AddToScheme).WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
 			if err := SetupIndexes(ctx, utiltesting.AsIndexer(kcBuilder)); err != nil {
 				t.Fatalf("Failed to setup indexes: %v", err)
 			}
diff --git a/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller_test.go b/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller_test.go
index ff1436a86..2111c441d 100644
--- a/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller_test.go
+++ b/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller_test.go
@@ -30,6 +30,7 @@ import (
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
@@ -578,7 +579,7 @@ func TestReconciler(t *testing.T) {
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
 			ctx, _ := utiltesting.ContextWithLog(t)
-			kcBuilder := utiltesting.NewClientBuilder(kftraining.AddToScheme)
+			kcBuilder := utiltesting.NewClientBuilder(kftraining.AddToScheme).WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
 			if err := SetupIndexes(ctx, utiltesting.AsIndexer(kcBuilder)); err != nil {
 				t.Fatalf("Failed to setup indexes: %v", err)
 			}
diff --git a/pkg/util/testing/wrappers.go b/pkg/util/testing/wrappers.go
index d20e4dbfe..7536e1897 100644
--- a/pkg/util/testing/wrappers.go
+++ b/pkg/util/testing/wrappers.go
@@ -394,6 +394,16 @@ func (w *WorkloadWrapper) SchedulingStatsEviction(evictionState kueue.WorkloadSc
 	return w
 }
 
+func (w *WorkloadWrapper) ClusterName(clusterName string) *WorkloadWrapper {
+	w.Status.ClusterName = &clusterName
+	return w
+}
+
+func (w *WorkloadWrapper) NominatedClusterNames(nominatedClusterNames []string) *WorkloadWrapper {
+	w.Status.NominatedClusterNames = nominatedClusterNames
+	return w
+}
+
 type PodSetWrapper struct{ kueue.PodSet }
 
 func MakePodSet(name kueue.PodSetReference, count int) *PodSetWrapper {
diff --git a/pkg/webhooks/workload_webhook_test.go b/pkg/webhooks/workload_webhook_test.go
index 4c7c46aa1..27a795ce9 100644
--- a/pkg/webhooks/workload_webhook_test.go
+++ b/pkg/webhooks/workload_webhook_test.go
@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/utils/ptr"
 
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/features"
 	testingutil "sigs.k8s.io/kueue/pkg/util/testing"
@@ -618,7 +619,7 @@ func TestValidateWorkloadUpdate(t *testing.T) {
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
 			features.SetFeatureGateDuringTest(t, features.ElasticJobsViaWorkloadSlices, tc.enableElasticJobsFeature)
-			errList := ValidateWorkloadUpdate(tc.after, tc.before)
+			errList := ValidateWorkloadUpdate(tc.after, tc.before, configapi.MultiKueueDispatcherModeAllAtOnce)
 			if diff := cmp.Diff(tc.wantErr, errList, cmpopts.IgnoreFields(field.Error{}, "Detail", "BadValue")); diff != "" {
 				t.Errorf("ValidateWorkloadUpdate() mismatch (-want +got):\n%s", diff)
 			}
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index 9f02aa191..0e07d26aa 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -38,6 +38,7 @@ 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"
@@ -958,6 +959,221 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 	})
 })
 
+var _ = ginkgo.Describe("MultiKueue with Incremental mode", ginkgo.Ordered, 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.BeforeAll(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.By("setting MultiKueue Dispatcher to Incremental", func() {
+			util.UpdateKueueConfiguration(ctx, k8sManagerClient, defaultManagerKueueCfg, managerClusterName, func(cfg *kueueconfig.Configuration) {
+				cfg.MultiKueue = &kueueconfig.MultiKueue{}
+				cfg.MultiKueue.DispatcherName = ptr.To(kueueconfig.MultiKueueDispatcherModeIncremental)
+			})
+		})
+	})
+
+	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.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.ExpectAllPodsInNamespaceDeleted(ctx, k8sManagerClient, managerNs)
+		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sWorker1Client, worker1Ns)
+		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sWorker2Client, worker2Ns)
+	})
+
+	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() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				workerWl := &kueue.Workload{}
+				g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
+				g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
+				workerJob := &batchv1.Job{}
+				g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(job), workerJob)).To(utiltesting.BeNotFoundError())
+				g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(job), workerJob)).To(utiltesting.BeNotFoundError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			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"))))
+		})
+	})
+})
+
 func waitForJobAdmitted(wlLookupKey types.NamespacedName, acName, workerName string) {
 	ginkgo.By(fmt.Sprintf("Waiting to be admitted in %s and manager", workerName))
 	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
diff --git a/test/e2e/multikueue/suite_test.go b/test/e2e/multikueue/suite_test.go
index 03699efd0..518e97e6e 100644
--- a/test/e2e/multikueue/suite_test.go
+++ b/test/e2e/multikueue/suite_test.go
@@ -43,6 +43,7 @@ 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"
@@ -67,6 +68,8 @@ var (
 	managerRestClient *rest.RESTClient
 	worker1RestClient *rest.RESTClient
 	worker2RestClient *rest.RESTClient
+
+	defaultManagerKueueCfg *v1beta1.Configuration
 )
 
 func policyRule(group, resource string, verbs ...string) rbacv1.PolicyRule {
@@ -317,6 +320,8 @@ 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() {
@@ -325,4 +330,7 @@ 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/multikueue/dispatcher_test.go b/test/integration/multikueue/dispatcher_test.go
new file mode 100644
index 000000000..97ca671ef
--- /dev/null
+++ b/test/integration/multikueue/dispatcher_test.go
@@ -0,0 +1,738 @@
+/*
+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 multikueue
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/clock"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/constants"
+	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
+	"sigs.k8s.io/kueue/pkg/features"
+	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("MultiKueueDispatcherIncremental", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		managerNs *corev1.Namespace
+		worker1Ns *corev1.Namespace
+		worker2Ns *corev1.Namespace
+
+		managerMultiKueueSecret1 *corev1.Secret
+		managerMultiKueueSecret2 *corev1.Secret
+		workerCluster1           *kueue.MultiKueueCluster
+		workerCluster2           *kueue.MultiKueueCluster
+		managerMultiKueueConfig  *kueue.MultiKueueConfig
+		multiKueueAC             *kueue.AdmissionCheck
+		managerCq                *kueue.ClusterQueue
+		managerLq                *kueue.LocalQueue
+
+		worker1Cq *kueue.ClusterQueue
+		worker1Lq *kueue.LocalQueue
+
+		worker2Cq *kueue.ClusterQueue
+		worker2Lq *kueue.LocalQueue
+		realClock = clock.RealClock{}
+	)
+
+	ginkgo.BeforeAll(func() {
+		managerTestCluster.fwk.StartManager(managerTestCluster.ctx, managerTestCluster.cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, defaultEnabledIntegrations, config.MultiKueueDispatcherModeIncremental)
+		})
+	})
+
+	ginkgo.AfterAll(func() {
+		managerTestCluster.fwk.StopManager(managerTestCluster.ctx)
+	})
+
+	ginkgo.BeforeEach(func() {
+		managerNs = util.CreateNamespaceFromPrefixWithLog(managerTestCluster.ctx, managerTestCluster.client, "multikueue-")
+		worker1Ns = util.CreateNamespaceWithLog(worker1TestCluster.ctx, worker1TestCluster.client, managerNs.Name)
+		worker2Ns = util.CreateNamespaceWithLog(worker2TestCluster.ctx, worker2TestCluster.client, managerNs.Name)
+
+		w1Kubeconfig, err := worker1TestCluster.kubeConfigBytes()
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		w2Kubeconfig, err := worker2TestCluster.kubeConfigBytes()
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		managerMultiKueueSecret1 = &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "multikueue1",
+				Namespace: managersConfigNamespace.Name,
+			},
+			Data: map[string][]byte{
+				kueue.MultiKueueConfigSecretKey: w1Kubeconfig,
+			},
+		}
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueSecret1)).To(gomega.Succeed())
+
+		managerMultiKueueSecret2 = &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "multikueue2",
+				Namespace: managersConfigNamespace.Name,
+			},
+			Data: map[string][]byte{
+				kueue.MultiKueueConfigSecretKey: w2Kubeconfig,
+			},
+		}
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueSecret2)).To(gomega.Succeed())
+
+		workerCluster1 = utiltesting.MakeMultiKueueCluster("worker1").KubeConfig(kueue.SecretLocationType, managerMultiKueueSecret1.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, workerCluster1)).To(gomega.Succeed())
+
+		workerCluster2 = utiltesting.MakeMultiKueueCluster("worker2").KubeConfig(kueue.SecretLocationType, managerMultiKueueSecret2.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, workerCluster2)).To(gomega.Succeed())
+
+		managerMultiKueueConfig = utiltesting.MakeMultiKueueConfig("multikueueconfig").Clusters(workerCluster1.Name, workerCluster2.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueConfig)).Should(gomega.Succeed())
+
+		multiKueueAC = utiltesting.MakeAdmissionCheck("ac1").
+			ControllerName(kueue.MultiKueueControllerName).
+			Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", managerMultiKueueConfig.Name).
+			Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, multiKueueAC)).Should(gomega.Succeed())
+
+		ginkgo.By("wait for check active", func() {
+			updatedAc := kueue.AdmissionCheck{}
+			acKey := client.ObjectKeyFromObject(multiKueueAC)
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, acKey, &updatedAc)).To(gomega.Succeed())
+				g.Expect(updatedAc.Status.Conditions).To(utiltesting.HaveConditionStatusTrue(kueue.AdmissionCheckActive))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		managerCq = utiltesting.MakeClusterQueue("q1").
+			AdmissionChecks(kueue.AdmissionCheckReference(multiKueueAC.Name)).
+			Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerCq)).Should(gomega.Succeed())
+
+		managerLq = utiltesting.MakeLocalQueue(managerCq.Name, managerNs.Name).ClusterQueue(managerCq.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerLq)).Should(gomega.Succeed())
+
+		worker1Cq = utiltesting.MakeClusterQueue("q1").Obj()
+		gomega.Expect(worker1TestCluster.client.Create(worker1TestCluster.ctx, worker1Cq)).Should(gomega.Succeed())
+		worker1Lq = utiltesting.MakeLocalQueue(worker1Cq.Name, worker1Ns.Name).ClusterQueue(worker1Cq.Name).Obj()
+		gomega.Expect(worker1TestCluster.client.Create(worker1TestCluster.ctx, worker1Lq)).Should(gomega.Succeed())
+
+		worker2Cq = utiltesting.MakeClusterQueue("q1").Obj()
+		gomega.Expect(worker2TestCluster.client.Create(worker2TestCluster.ctx, worker2Cq)).Should(gomega.Succeed())
+		worker2Lq = utiltesting.MakeLocalQueue(worker2Cq.Name, worker2Ns.Name).ClusterQueue(worker2Cq.Name).Obj()
+		gomega.Expect(worker2TestCluster.client.Create(worker2TestCluster.ctx, worker2Lq)).Should(gomega.Succeed())
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(managerTestCluster.ctx, managerTestCluster.client, managerNs)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(worker1TestCluster.ctx, worker1TestCluster.client, worker1Ns)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(worker2TestCluster.ctx, worker2TestCluster.client, worker2Ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerCq, true)
+		util.ExpectObjectToBeDeleted(worker1TestCluster.ctx, worker1TestCluster.client, worker1Cq, true)
+		util.ExpectObjectToBeDeleted(worker2TestCluster.ctx, worker2TestCluster.client, worker2Cq, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, multiKueueAC, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueConfig, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, workerCluster1, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, workerCluster2, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueSecret1, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueSecret2, true)
+	})
+
+	ginkgo.It("Should run a job on worker if admitted (ManagedBy)", func() {
+		features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.MultiKueueBatchJobWithManagedBy, true)
+		job := testingjob.MakeJob("job", managerNs.Name).
+			ManagedBy(kueue.MultiKueueControllerName).
+			Queue(kueue.LocalQueueName(managerLq.Name)).
+			Obj()
+		util.MustCreate(managerTestCluster.ctx, managerTestCluster.client, job)
+
+		createdWorkload := &kueue.Workload{}
+		wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: managerNs.Name}
+
+		ginkgo.By("setting workload reservation in the management cluster", func() {
+			admission := utiltesting.MakeAdmission(managerCq.Name).Obj()
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(managerTestCluster.ctx, managerTestCluster.client, createdWorkload, admission)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload creation in the worker clusters 1 and 2", func() {
+			managerWl := &kueue.Workload{}
+			gomega.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec))
+				// The workload should be created in worker2 as well, since the job is managed
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec))
+				// nominated workers should be updated in the manager
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+				ginkgo.GinkgoLogr.Info(fmt.Sprintf("Workload status in manager: %s, %v", managerWl.Status.NominatedClusterNames, managerWl.Status.Conditions))
+				g.Expect(managerWl.Status.NominatedClusterNames).To(gomega.ContainElements(workerCluster1.Name, workerCluster2.Name))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("setting workload reservation in worker1, workload in worker2 is removed", func() {
+			admission := utiltesting.MakeAdmission(managerCq.Name).Obj()
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(worker1TestCluster.ctx, worker1TestCluster.client, createdWorkload, admission)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(utiltesting.BeNotFoundError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload ClusterName in the management cluster", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.ClusterName).To(gomega.HaveValue(gomega.Equal(workerCluster1.Name)))
+				g.Expect(createdWorkload.Status.NominatedClusterNames).To(gomega.BeEmpty())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("setting the check conditions for eviction", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				wlPatch := workload.PrepareWorkloadPatch(createdWorkload, true, realClock)
+				workload.SetAdmissionCheckState(&wlPatch.Status.AdmissionChecks, kueue.AdmissionCheckState{
+					Name:    kueue.AdmissionCheckReference(multiKueueAC.Name),
+					State:   kueue.CheckStateRejected,
+					Message: "check rejected",
+				}, realClock)
+				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload ClusterName and NominatedClusterNames are reset in the management cluster", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.NominatedClusterNames).To(gomega.BeNil())
+				g.Expect(createdWorkload.Status.ClusterName).To(gomega.BeNil())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
+})
+
+var _ = ginkgo.Describe("MultiKueueDispatcherExternal", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		managerNs *corev1.Namespace
+		worker1Ns *corev1.Namespace
+		worker2Ns *corev1.Namespace
+
+		managerMultiKueueSecret1 *corev1.Secret
+		managerMultiKueueSecret2 *corev1.Secret
+		workerCluster1           *kueue.MultiKueueCluster
+		workerCluster2           *kueue.MultiKueueCluster
+		managerMultiKueueConfig  *kueue.MultiKueueConfig
+		multiKueueAC             *kueue.AdmissionCheck
+		managerCq                *kueue.ClusterQueue
+		managerLq                *kueue.LocalQueue
+		realClock                = clock.RealClock{}
+
+		worker1Cq *kueue.ClusterQueue
+		worker1Lq *kueue.LocalQueue
+
+		worker2Cq *kueue.ClusterQueue
+		worker2Lq *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeAll(func() {
+		managerTestCluster.fwk.StartManager(managerTestCluster.ctx, managerTestCluster.cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, defaultEnabledIntegrations, "example.com/custom-mk-dispatcher")
+		})
+	})
+
+	ginkgo.AfterAll(func() {
+		managerTestCluster.fwk.StopManager(managerTestCluster.ctx)
+	})
+
+	ginkgo.BeforeEach(func() {
+		managerNs = util.CreateNamespaceFromPrefixWithLog(managerTestCluster.ctx, managerTestCluster.client, "multikueue-")
+		worker1Ns = util.CreateNamespaceWithLog(worker1TestCluster.ctx, worker1TestCluster.client, managerNs.Name)
+		worker2Ns = util.CreateNamespaceWithLog(worker2TestCluster.ctx, worker2TestCluster.client, managerNs.Name)
+
+		w1Kubeconfig, err := worker1TestCluster.kubeConfigBytes()
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		w2Kubeconfig, err := worker2TestCluster.kubeConfigBytes()
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		managerMultiKueueSecret1 = &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "multikueue1",
+				Namespace: managersConfigNamespace.Name,
+			},
+			Data: map[string][]byte{
+				kueue.MultiKueueConfigSecretKey: w1Kubeconfig,
+			},
+		}
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueSecret1)).To(gomega.Succeed())
+
+		managerMultiKueueSecret2 = &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "multikueue2",
+				Namespace: managersConfigNamespace.Name,
+			},
+			Data: map[string][]byte{
+				kueue.MultiKueueConfigSecretKey: w2Kubeconfig,
+			},
+		}
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueSecret2)).To(gomega.Succeed())
+
+		workerCluster1 = utiltesting.MakeMultiKueueCluster("worker1").KubeConfig(kueue.SecretLocationType, managerMultiKueueSecret1.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, workerCluster1)).To(gomega.Succeed())
+
+		workerCluster2 = utiltesting.MakeMultiKueueCluster("worker2").KubeConfig(kueue.SecretLocationType, managerMultiKueueSecret2.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, workerCluster2)).To(gomega.Succeed())
+
+		managerMultiKueueConfig = utiltesting.MakeMultiKueueConfig("multikueueconfig").Clusters(workerCluster1.Name, workerCluster2.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueConfig)).Should(gomega.Succeed())
+
+		multiKueueAC = utiltesting.MakeAdmissionCheck("ac1").
+			ControllerName(kueue.MultiKueueControllerName).
+			Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", managerMultiKueueConfig.Name).
+			Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, multiKueueAC)).Should(gomega.Succeed())
+
+		ginkgo.By("wait for check active", func() {
+			updatedAc := kueue.AdmissionCheck{}
+			acKey := client.ObjectKeyFromObject(multiKueueAC)
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, acKey, &updatedAc)).To(gomega.Succeed())
+				g.Expect(updatedAc.Status.Conditions).To(utiltesting.HaveConditionStatusTrue(kueue.AdmissionCheckActive))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		managerCq = utiltesting.MakeClusterQueue("q1").
+			AdmissionChecks(kueue.AdmissionCheckReference(multiKueueAC.Name)).
+			Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerCq)).Should(gomega.Succeed())
+
+		managerLq = utiltesting.MakeLocalQueue(managerCq.Name, managerNs.Name).ClusterQueue(managerCq.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerLq)).Should(gomega.Succeed())
+
+		worker1Cq = utiltesting.MakeClusterQueue("q1").Obj()
+		gomega.Expect(worker1TestCluster.client.Create(worker1TestCluster.ctx, worker1Cq)).Should(gomega.Succeed())
+		worker1Lq = utiltesting.MakeLocalQueue(worker1Cq.Name, worker1Ns.Name).ClusterQueue(worker1Cq.Name).Obj()
+		gomega.Expect(worker1TestCluster.client.Create(worker1TestCluster.ctx, worker1Lq)).Should(gomega.Succeed())
+
+		worker2Cq = utiltesting.MakeClusterQueue("q1").Obj()
+		gomega.Expect(worker2TestCluster.client.Create(worker2TestCluster.ctx, worker2Cq)).Should(gomega.Succeed())
+		worker2Lq = utiltesting.MakeLocalQueue(worker2Cq.Name, worker2Ns.Name).ClusterQueue(worker2Cq.Name).Obj()
+		gomega.Expect(worker2TestCluster.client.Create(worker2TestCluster.ctx, worker2Lq)).Should(gomega.Succeed())
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(managerTestCluster.ctx, managerTestCluster.client, managerNs)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(worker1TestCluster.ctx, worker1TestCluster.client, worker1Ns)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(worker2TestCluster.ctx, worker2TestCluster.client, worker2Ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerCq, true)
+		util.ExpectObjectToBeDeleted(worker1TestCluster.ctx, worker1TestCluster.client, worker1Cq, true)
+		util.ExpectObjectToBeDeleted(worker2TestCluster.ctx, worker2TestCluster.client, worker2Cq, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, multiKueueAC, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueConfig, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, workerCluster1, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, workerCluster2, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueSecret1, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueSecret2, true)
+	})
+
+	ginkgo.It("Should run a job on worker if admitted (ManagedBy)", func() {
+		features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.MultiKueueBatchJobWithManagedBy, true)
+		job := testingjob.MakeJob("job", managerNs.Name).
+			ManagedBy(kueue.MultiKueueControllerName).
+			Queue(kueue.LocalQueueName(managerLq.Name)).
+			Obj()
+		util.MustCreate(managerTestCluster.ctx, managerTestCluster.client, job)
+
+		createdWorkload := &kueue.Workload{}
+		wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: managerNs.Name}
+
+		ginkgo.By("setting workload reservation in the management cluster", func() {
+			admission := utiltesting.MakeAdmission(managerCq.Name).Obj()
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(managerTestCluster.ctx, managerTestCluster.client, createdWorkload, admission)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload was not created in any worker cluster", func() {
+			managerWl := &kueue.Workload{}
+			gomega.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+			gomega.Expect(managerWl.Status.ClusterName).To(gomega.BeNil())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(utiltesting.BeNotFoundError())
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(utiltesting.BeNotFoundError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("updating workload nomination by cluster2", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				managerWl := &kueue.Workload{}
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+				wlPatch := workload.PrepareWorkloadPatch(managerWl, true, realClock)
+				wlPatch.Status.NominatedClusterNames = []string{workerCluster2.Name}
+				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload creation in the worker clusters 2", func() {
+			managerWl := &kueue.Workload{}
+			gomega.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(utiltesting.BeNotFoundError())
+
+				// The workload should be only created in worker2
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec))
+
+				// nominated workers should be updated in the manager
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+				ginkgo.GinkgoLogr.Info(fmt.Sprintf("Workload status in manager: %s, %v", managerWl.Status.NominatedClusterNames, managerWl.Status.Conditions))
+				g.Expect(managerWl.Status.NominatedClusterNames).To(gomega.ContainElements(workerCluster2.Name))
+				g.Expect(managerWl.Status.ClusterName).To(gomega.BeNil())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("updating workload nomination by cluster1 and cluster2", func() {
+			managerWl := &kueue.Workload{}
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+				wlPatch := workload.PrepareWorkloadPatch(managerWl, true, realClock)
+				wlPatch.Status.NominatedClusterNames = []string{workerCluster1.Name, workerCluster2.Name}
+				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload creation in the worker clusters 1 and 2", func() {
+			managerWl := &kueue.Workload{}
+			gomega.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec))
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec))
+				// nominated workers should be updated in the manager
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+				ginkgo.GinkgoLogr.Info(fmt.Sprintf("Workload status in manager: %s, %v", managerWl.Status.NominatedClusterNames, managerWl.Status.Conditions))
+				g.Expect(managerWl.Status.NominatedClusterNames).To(gomega.ContainElements(workerCluster1.Name, workerCluster2.Name))
+				g.Expect(managerWl.Status.ClusterName).To(gomega.BeNil())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("setting workload reservation in worker2, workload in worker1 is removed", func() {
+			admission := utiltesting.MakeAdmission(managerCq.Name).Obj()
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(worker2TestCluster.ctx, worker2TestCluster.client, createdWorkload, admission)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(utiltesting.BeNotFoundError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload ClusterName in the management cluster", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.ClusterName).To(gomega.HaveValue(gomega.Equal(workerCluster2.Name)))
+				g.Expect(createdWorkload.Status.NominatedClusterNames).To(gomega.BeEmpty())
+			}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("setting the check conditions for eviction", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				wlPatch := workload.PrepareWorkloadPatch(createdWorkload, true, realClock)
+				workload.SetAdmissionCheckState(&wlPatch.Status.AdmissionChecks, kueue.AdmissionCheckState{
+					Name:    kueue.AdmissionCheckReference(multiKueueAC.Name),
+					State:   kueue.CheckStateRejected,
+					Message: "check rejected",
+				}, realClock)
+				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload ClusterName and NominatedClusterNames are reset in the management cluster", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.NominatedClusterNames).To(gomega.BeNil())
+				g.Expect(createdWorkload.Status.ClusterName).To(gomega.BeNil())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
+})
+
+var _ = ginkgo.Describe("MultiKueueDispatcherAllAtOnce", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		managerNs *corev1.Namespace
+		worker1Ns *corev1.Namespace
+		worker2Ns *corev1.Namespace
+
+		managerMultiKueueSecret1 *corev1.Secret
+		managerMultiKueueSecret2 *corev1.Secret
+		workerCluster1           *kueue.MultiKueueCluster
+		workerCluster2           *kueue.MultiKueueCluster
+		managerMultiKueueConfig  *kueue.MultiKueueConfig
+		multiKueueAC             *kueue.AdmissionCheck
+		managerCq                *kueue.ClusterQueue
+		managerLq                *kueue.LocalQueue
+
+		worker1Cq *kueue.ClusterQueue
+		worker1Lq *kueue.LocalQueue
+
+		worker2Cq *kueue.ClusterQueue
+		worker2Lq *kueue.LocalQueue
+		realClock = clock.RealClock{}
+	)
+
+	ginkgo.BeforeAll(func() {
+		managerTestCluster.fwk.StartManager(managerTestCluster.ctx, managerTestCluster.cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, defaultEnabledIntegrations, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+
+	ginkgo.AfterAll(func() {
+		managerTestCluster.fwk.StopManager(managerTestCluster.ctx)
+	})
+
+	ginkgo.BeforeEach(func() {
+		managerNs = util.CreateNamespaceFromPrefixWithLog(managerTestCluster.ctx, managerTestCluster.client, "multikueue-")
+		worker1Ns = util.CreateNamespaceWithLog(worker1TestCluster.ctx, worker1TestCluster.client, managerNs.Name)
+		worker2Ns = util.CreateNamespaceWithLog(worker2TestCluster.ctx, worker2TestCluster.client, managerNs.Name)
+
+		w1Kubeconfig, err := worker1TestCluster.kubeConfigBytes()
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		w2Kubeconfig, err := worker2TestCluster.kubeConfigBytes()
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		managerMultiKueueSecret1 = &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "multikueue1",
+				Namespace: managersConfigNamespace.Name,
+			},
+			Data: map[string][]byte{
+				kueue.MultiKueueConfigSecretKey: w1Kubeconfig,
+			},
+		}
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueSecret1)).To(gomega.Succeed())
+
+		managerMultiKueueSecret2 = &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "multikueue2",
+				Namespace: managersConfigNamespace.Name,
+			},
+			Data: map[string][]byte{
+				kueue.MultiKueueConfigSecretKey: w2Kubeconfig,
+			},
+		}
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueSecret2)).To(gomega.Succeed())
+
+		workerCluster1 = utiltesting.MakeMultiKueueCluster("worker1").KubeConfig(kueue.SecretLocationType, managerMultiKueueSecret1.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, workerCluster1)).To(gomega.Succeed())
+
+		workerCluster2 = utiltesting.MakeMultiKueueCluster("worker2").KubeConfig(kueue.SecretLocationType, managerMultiKueueSecret2.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, workerCluster2)).To(gomega.Succeed())
+
+		managerMultiKueueConfig = utiltesting.MakeMultiKueueConfig("multikueueconfig").Clusters(workerCluster1.Name, workerCluster2.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerMultiKueueConfig)).Should(gomega.Succeed())
+
+		multiKueueAC = utiltesting.MakeAdmissionCheck("ac1").
+			ControllerName(kueue.MultiKueueControllerName).
+			Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", managerMultiKueueConfig.Name).
+			Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, multiKueueAC)).Should(gomega.Succeed())
+
+		ginkgo.By("wait for check active", func() {
+			updatedAc := kueue.AdmissionCheck{}
+			acKey := client.ObjectKeyFromObject(multiKueueAC)
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, acKey, &updatedAc)).To(gomega.Succeed())
+				g.Expect(updatedAc.Status.Conditions).To(utiltesting.HaveConditionStatusTrue(kueue.AdmissionCheckActive))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		managerCq = utiltesting.MakeClusterQueue("q1").
+			AdmissionChecks(kueue.AdmissionCheckReference(multiKueueAC.Name)).
+			Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerCq)).Should(gomega.Succeed())
+
+		managerLq = utiltesting.MakeLocalQueue(managerCq.Name, managerNs.Name).ClusterQueue(managerCq.Name).Obj()
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, managerLq)).Should(gomega.Succeed())
+
+		worker1Cq = utiltesting.MakeClusterQueue("q1").Obj()
+		gomega.Expect(worker1TestCluster.client.Create(worker1TestCluster.ctx, worker1Cq)).Should(gomega.Succeed())
+		worker1Lq = utiltesting.MakeLocalQueue(worker1Cq.Name, worker1Ns.Name).ClusterQueue(worker1Cq.Name).Obj()
+		gomega.Expect(worker1TestCluster.client.Create(worker1TestCluster.ctx, worker1Lq)).Should(gomega.Succeed())
+
+		worker2Cq = utiltesting.MakeClusterQueue("q1").Obj()
+		gomega.Expect(worker2TestCluster.client.Create(worker2TestCluster.ctx, worker2Cq)).Should(gomega.Succeed())
+		worker2Lq = utiltesting.MakeLocalQueue(worker2Cq.Name, worker2Ns.Name).ClusterQueue(worker2Cq.Name).Obj()
+		gomega.Expect(worker2TestCluster.client.Create(worker2TestCluster.ctx, worker2Lq)).Should(gomega.Succeed())
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(managerTestCluster.ctx, managerTestCluster.client, managerNs)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(worker1TestCluster.ctx, worker1TestCluster.client, worker1Ns)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(worker2TestCluster.ctx, worker2TestCluster.client, worker2Ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerCq, true)
+		util.ExpectObjectToBeDeleted(worker1TestCluster.ctx, worker1TestCluster.client, worker1Cq, true)
+		util.ExpectObjectToBeDeleted(worker2TestCluster.ctx, worker2TestCluster.client, worker2Cq, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, multiKueueAC, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueConfig, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, workerCluster1, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, workerCluster2, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueSecret1, true)
+		util.ExpectObjectToBeDeleted(managerTestCluster.ctx, managerTestCluster.client, managerMultiKueueSecret2, true)
+	})
+
+	ginkgo.It("Should run a job on worker if admitted after the upgrade to MultiKueue Dispatcher", func() {
+		features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.MultiKueueBatchJobWithManagedBy, true)
+		job := testingjob.MakeJob("job", managerNs.Name).
+			ManagedBy(kueue.MultiKueueControllerName).
+			Queue(kueue.LocalQueueName(managerLq.Name)).
+			Obj()
+		util.MustCreate(managerTestCluster.ctx, managerTestCluster.client, job)
+
+		createdWorkload := &kueue.Workload{}
+		wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: managerNs.Name}
+
+		ginkgo.By("setting workload reservation in the management cluster", func() {
+			admission := utiltesting.MakeAdmission(managerCq.Name).Obj()
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(managerTestCluster.ctx, managerTestCluster.client, createdWorkload, admission)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload creation in the worker clusters 1 and 2", func() {
+			managerWl := &kueue.Workload{}
+			gomega.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec))
+				// The workload should be created in worker2 as well, since the job is managed
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec))
+				// nominated workers should be updated in the manager
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, managerWl)).To(gomega.Succeed())
+				ginkgo.GinkgoLogr.Info(fmt.Sprintf("Workload status in manager: %s, %v", managerWl.Status.NominatedClusterNames, managerWl.Status.Conditions))
+				g.Expect(managerWl.Status.NominatedClusterNames).To(gomega.ContainElements(workerCluster1.Name, workerCluster2.Name))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("setting workload reservation in worker1, workload in worker2 is removed", func() {
+			admission := utiltesting.MakeAdmission(managerCq.Name).Obj()
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker1TestCluster.client.Get(worker1TestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(worker1TestCluster.ctx, worker1TestCluster.client, createdWorkload, admission)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, wlLookupKey, createdWorkload)).To(utiltesting.BeNotFoundError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking workload status conditions", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.Conditions).To(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadQuotaReserved,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadQuotaReserved,
+						Message: fmt.Sprintf("Quota reserved in ClusterQueue %s", managerLq.Name),
+					}, util.IgnoreConditionTimestampsAndObservedGeneration),
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadAdmitted,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadAdmitted,
+						Message: "The workload is admitted",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration),
+				))
+				state := workload.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAC.Name))
+				g.Expect(state).NotTo(gomega.BeNil())
+				g.Expect(state.State).To(gomega.Equal(kueue.CheckStateReady))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload ClusterName in the management cluster", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.ClusterName).To(gomega.HaveValue(gomega.Equal(workerCluster1.Name)))
+				g.Expect(createdWorkload.Status.NominatedClusterNames).To(gomega.BeEmpty())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("Resetting ClusterName field in the manager", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				wlPatch := workload.PrepareWorkloadPatch(createdWorkload, true, realClock)
+				wlPatch.Status.ClusterName = nil
+				g.Expect(managerTestCluster.client.Status().Patch(managerTestCluster.ctx, wlPatch, client.Apply, client.FieldOwner(constants.AdmissionName), client.ForceOwnership)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking the workload ClusterName and NominatedClusterNames are reset in the management cluster", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.NominatedClusterNames).To(gomega.BeNil())
+				g.Expect(createdWorkload.Status.ClusterName).To(gomega.BeNil())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("checking workload status conditions again", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+				g.Expect(createdWorkload.Status.Conditions).To(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadQuotaReserved,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadQuotaReserved,
+						Message: fmt.Sprintf("Quota reserved in ClusterQueue %s", managerLq.Name),
+					}, util.IgnoreConditionTimestampsAndObservedGeneration),
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadAdmitted,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadAdmitted,
+						Message: "The workload is admitted",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration),
+				))
+				state := workload.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAC.Name))
+				g.Expect(state).NotTo(gomega.BeNil())
+				g.Expect(state.State).To(gomega.Equal(kueue.CheckStateReady))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		time.Sleep(5 * time.Second)
+	})
+})
diff --git a/test/integration/multikueue/enabled_integration_test.go b/test/integration/multikueue/enabled_integration_test.go
index 4ecaa0a61..369490ed5 100644
--- a/test/integration/multikueue/enabled_integration_test.go
+++ b/test/integration/multikueue/enabled_integration_test.go
@@ -35,6 +35,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	workloadmpijob "sigs.k8s.io/kueue/pkg/controller/jobs/mpijob"
@@ -63,7 +64,7 @@ var _ = ginkgo.Describe("MultiKueue when not all integrations are enabled", gink
 
 	ginkgo.BeforeAll(func() {
 		managerTestCluster.fwk.StartManager(managerTestCluster.ctx, managerTestCluster.cfg, func(ctx context.Context, mgr manager.Manager) {
-			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, sets.New("batch/job"))
+			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, sets.New("batch/job"), config.MultiKueueDispatcherModeAllAtOnce)
 		})
 	})
 
diff --git a/test/integration/multikueue/jobs_test.go b/test/integration/multikueue/jobs_test.go
index b44bf6854..0a7a26520 100644
--- a/test/integration/multikueue/jobs_test.go
+++ b/test/integration/multikueue/jobs_test.go
@@ -39,6 +39,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	jobset "sigs.k8s.io/jobset/api/jobset/v1alpha2"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	workloadappwrapper "sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
@@ -99,7 +100,7 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 
 	ginkgo.BeforeAll(func() {
 		managerTestCluster.fwk.StartManager(managerTestCluster.ctx, managerTestCluster.cfg, func(ctx context.Context, mgr manager.Manager) {
-			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, defaultEnabledIntegrations)
+			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, defaultEnabledIntegrations, config.MultiKueueDispatcherModeAllAtOnce)
 		})
 	})
 
diff --git a/test/integration/multikueue/no_gc_test.go b/test/integration/multikueue/no_gc_test.go
index 0501ae84c..1df02ffe5 100644
--- a/test/integration/multikueue/no_gc_test.go
+++ b/test/integration/multikueue/no_gc_test.go
@@ -29,6 +29,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -64,7 +65,7 @@ var _ = ginkgo.Describe("MultiKueue no GC", ginkgo.Ordered, ginkgo.ContinueOnFai
 
 	ginkgo.BeforeAll(func() {
 		managerTestCluster.fwk.StartManager(managerTestCluster.ctx, managerTestCluster.cfg, func(ctx context.Context, mgr manager.Manager) {
-			managerAndMultiKueueSetup(ctx, mgr, 0, defaultEnabledIntegrations)
+			managerAndMultiKueueSetup(ctx, mgr, 0, defaultEnabledIntegrations, config.MultiKueueDispatcherModeAllAtOnce)
 		})
 	})
 
diff --git a/test/integration/multikueue/setup_test.go b/test/integration/multikueue/setup_test.go
index a8ed8687f..825f9e9b6 100644
--- a/test/integration/multikueue/setup_test.go
+++ b/test/integration/multikueue/setup_test.go
@@ -30,6 +30,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/test/util"
@@ -59,7 +60,7 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 
 	ginkgo.BeforeAll(func() {
 		managerTestCluster.fwk.StartManager(managerTestCluster.ctx, managerTestCluster.cfg, func(ctx context.Context, mgr manager.Manager) {
-			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, defaultEnabledIntegrations)
+			managerAndMultiKueueSetup(ctx, mgr, 2*time.Second, defaultEnabledIntegrations, config.MultiKueueDispatcherModeAllAtOnce)
 		})
 	})
 
diff --git a/test/integration/multikueue/suite_test.go b/test/integration/multikueue/suite_test.go
index 167614f23..0d915db59 100644
--- a/test/integration/multikueue/suite_test.go
+++ b/test/integration/multikueue/suite_test.go
@@ -30,6 +30,7 @@ import (
 	versionutil "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/rest"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
@@ -125,7 +126,7 @@ func managerSetup(ctx context.Context, mgr manager.Manager) {
 	failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
 
-	failedWebhook, err := webhooks.Setup(mgr)
+	failedWebhook, err := webhooks.Setup(mgr, ptr.Deref(configuration.MultiKueue.DispatcherName, config.MultiKueueDispatcherModeAllAtOnce))
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 	err = workloadjob.SetupIndexes(ctx, mgr.GetFieldIndexer())
@@ -273,7 +274,13 @@ func managerSetup(ctx context.Context, mgr manager.Manager) {
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 }
 
-func managerAndMultiKueueSetup(ctx context.Context, mgr manager.Manager, gcInterval time.Duration, enabledIntegrations sets.Set[string]) {
+func managerAndMultiKueueSetup(
+	ctx context.Context,
+	mgr manager.Manager,
+	gcInterval time.Duration,
+	enabledIntegrations sets.Set[string],
+	dispatcherName string,
+) {
 	managerSetup(ctx, mgr)
 
 	err := multikueue.SetupIndexer(ctx, mgr.GetFieldIndexer(), managersConfigNamespace.Name)
@@ -287,6 +294,7 @@ func managerAndMultiKueueSetup(ctx context.Context, mgr manager.Manager, gcInter
 		multikueue.WithWorkerLostTimeout(testingWorkerLostTimeout),
 		multikueue.WithEventsBatchPeriod(100*time.Millisecond),
 		multikueue.WithAdapters(adapters),
+		multikueue.WithDispatcherName(dispatcherName),
 	)
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 }
diff --git a/test/integration/singlecluster/controller/admissionchecks/provisioning/suite_test.go b/test/integration/singlecluster/controller/admissionchecks/provisioning/suite_test.go
index b546ec9e7..96bc817a2 100644
--- a/test/integration/singlecluster/controller/admissionchecks/provisioning/suite_test.go
+++ b/test/integration/singlecluster/controller/admissionchecks/provisioning/suite_test.go
@@ -72,7 +72,7 @@ func managerSetup() framework.ManagerSetup {
 		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
-		failedWebhook, err := webhooks.Setup(mgr)
+		failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 		controllersCfg := &config.Configuration{}
diff --git a/test/integration/singlecluster/controller/core/suite_test.go b/test/integration/singlecluster/controller/core/suite_test.go
index 599bcb806..7639ed5df 100644
--- a/test/integration/singlecluster/controller/core/suite_test.go
+++ b/test/integration/singlecluster/controller/core/suite_test.go
@@ -72,7 +72,7 @@ func managerAndControllerSetup(controllersCfg *config.Configuration) framework.M
 		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
-		failedWebhook, err := webhooks.Setup(mgr)
+		failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 		if controllersCfg == nil {
diff --git a/test/integration/singlecluster/controller/jobs/pod/suite_test.go b/test/integration/singlecluster/controller/jobs/pod/suite_test.go
index 1adf77e55..16afb4147 100644
--- a/test/integration/singlecluster/controller/jobs/pod/suite_test.go
+++ b/test/integration/singlecluster/controller/jobs/pod/suite_test.go
@@ -127,7 +127,7 @@ func managerSetup(
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 		err = pod.SetupWebhook(mgr, opts...)
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
-		failedWebhook, err := webhooks.Setup(mgr)
+		failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 		if setupTASControllers {
diff --git a/test/integration/singlecluster/importer/suite_test.go b/test/integration/singlecluster/importer/suite_test.go
index 50cae9cf4..5126d6481 100644
--- a/test/integration/singlecluster/importer/suite_test.go
+++ b/test/integration/singlecluster/importer/suite_test.go
@@ -77,7 +77,7 @@ func managerAndSchedulerSetup(ctx context.Context, mgr manager.Manager) {
 	failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
 
-	failedWebhook, err := webhooks.Setup(mgr)
+	failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 	err = workloadjob.SetupIndexes(ctx, mgr.GetFieldIndexer())
diff --git a/test/integration/singlecluster/kueuectl/suite_test.go b/test/integration/singlecluster/kueuectl/suite_test.go
index c26df9600..73e798a4c 100644
--- a/test/integration/singlecluster/kueuectl/suite_test.go
+++ b/test/integration/singlecluster/kueuectl/suite_test.go
@@ -79,7 +79,7 @@ func managerSetup(ctx context.Context, mgr manager.Manager) {
 	err := indexer.Setup(ctx, mgr.GetFieldIndexer())
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
-	failedWebhook, err := webhooks.Setup(mgr)
+	failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 	controllersCfg := &config.Configuration{}
diff --git a/test/integration/singlecluster/scheduler/fairsharing/suite_test.go b/test/integration/singlecluster/scheduler/fairsharing/suite_test.go
index 75e2ea241..c9fe0dd66 100644
--- a/test/integration/singlecluster/scheduler/fairsharing/suite_test.go
+++ b/test/integration/singlecluster/scheduler/fairsharing/suite_test.go
@@ -76,10 +76,10 @@ func managerAndSchedulerSetup(ctx context.Context, mgr manager.Manager) {
 	}
 	admissionFairSharing := &config.AdmissionFairSharing{
 		UsageHalfLifeTime: metav1.Duration{
-			Duration: 3 * time.Second,
+			Duration: 10 * time.Second,
 		},
 		UsageSamplingInterval: metav1.Duration{
-			Duration: 3 * time.Second,
+			Duration: 1 * time.Second,
 		},
 	}
 
@@ -97,7 +97,7 @@ func managerAndSchedulerSetup(ctx context.Context, mgr manager.Manager) {
 	failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
 
-	failedWebhook, err := webhooks.Setup(mgr)
+	failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 	err = workloadjob.SetupIndexes(ctx, mgr.GetFieldIndexer())
diff --git a/test/integration/singlecluster/scheduler/podsready/suite_test.go b/test/integration/singlecluster/scheduler/podsready/suite_test.go
index 3498355ee..7880fe992 100644
--- a/test/integration/singlecluster/scheduler/podsready/suite_test.go
+++ b/test/integration/singlecluster/scheduler/podsready/suite_test.go
@@ -101,7 +101,7 @@ func managerAndSchedulerSetup(configuration *config.Configuration) framework.Man
 		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
 
-		failedWebhook, err := webhooks.Setup(mgr)
+		failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 		err = workloadjob.SetupIndexes(ctx, mgr.GetFieldIndexer())
diff --git a/test/integration/singlecluster/scheduler/suite_test.go b/test/integration/singlecluster/scheduler/suite_test.go
index 470587d09..5b8b10dc4 100644
--- a/test/integration/singlecluster/scheduler/suite_test.go
+++ b/test/integration/singlecluster/scheduler/suite_test.go
@@ -90,7 +90,7 @@ func managerAndSchedulerSetup(ctx context.Context, mgr manager.Manager) {
 	failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
 
-	failedWebhook, err := webhooks.Setup(mgr)
+	failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 	err = workloadjob.SetupIndexes(ctx, mgr.GetFieldIndexer())
diff --git a/test/integration/singlecluster/tas/suite_test.go b/test/integration/singlecluster/tas/suite_test.go
index 2983948be..25d46d34b 100644
--- a/test/integration/singlecluster/tas/suite_test.go
+++ b/test/integration/singlecluster/tas/suite_test.go
@@ -75,7 +75,7 @@ func managerSetup(ctx context.Context, mgr manager.Manager) {
 	err := indexer.Setup(ctx, mgr.GetFieldIndexer())
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
-	failedWebhook, err := webhooks.Setup(mgr)
+	failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 	controllersCfg := &config.Configuration{}
diff --git a/test/integration/singlecluster/webhook/core/admissioncheck_test.go b/test/integration/singlecluster/webhook/core/admissioncheck_test.go
index 37e864174..97596f5fe 100644
--- a/test/integration/singlecluster/webhook/core/admissioncheck_test.go
+++ b/test/integration/singlecluster/webhook/core/admissioncheck_test.go
@@ -17,6 +17,8 @@ limitations under the License.
 package core
 
 import (
+	"context"
+
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
@@ -24,13 +26,23 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/test/util"
 )
 
-var _ = ginkgo.Describe("AdmissionCheck Webhook", func() {
+var _ = ginkgo.Describe("AdmissionCheck Webhook", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
 	ginkgo.When("Creating a AdmissionCheck", func() {
 		ginkgo.DescribeTable("Defaulting on creating", func(ac, wantAC kueue.AdmissionCheck) {
 			util.MustCreate(ctx, k8sClient, &ac)
diff --git a/test/integration/singlecluster/webhook/core/clusterqueue_test.go b/test/integration/singlecluster/webhook/core/clusterqueue_test.go
index 431ba9cc0..60d58444e 100644
--- a/test/integration/singlecluster/webhook/core/clusterqueue_test.go
+++ b/test/integration/singlecluster/webhook/core/clusterqueue_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package core
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -28,7 +29,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/test/util"
@@ -39,7 +42,15 @@ const (
 	flavorsMaxItems   = 16
 )
 
-var _ = ginkgo.Describe("ClusterQueue Webhook", func() {
+var _ = ginkgo.Describe("ClusterQueue Webhook", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
 	var ns *corev1.Namespace
 	defaultFlavorFungibility := &kueue.FlavorFungibility{
 		WhenCanBorrow:  kueue.Borrow,
diff --git a/test/integration/singlecluster/webhook/core/cohort_test.go b/test/integration/singlecluster/webhook/core/cohort_test.go
index 12b1c5022..228daee9a 100644
--- a/test/integration/singlecluster/webhook/core/cohort_test.go
+++ b/test/integration/singlecluster/webhook/core/cohort_test.go
@@ -17,6 +17,8 @@ limitations under the License.
 package core
 
 import (
+	"context"
+
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	"github.com/onsi/gomega/types"
@@ -24,13 +26,23 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/test/util"
 )
 
-var _ = ginkgo.Describe("Cohort Webhook", func() {
+var _ = ginkgo.Describe("Cohort Webhook", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
 	ginkgo.When("Creating a Cohort", func() {
 		ginkgo.DescribeTable("Validate Cohort on creation", func(cohort *kueue.Cohort, matcher types.GomegaMatcher) {
 			err := k8sClient.Create(ctx, cohort)
diff --git a/test/integration/singlecluster/webhook/core/localqueue_test.go b/test/integration/singlecluster/webhook/core/localqueue_test.go
index 292802d9f..f4cdbfbe0 100644
--- a/test/integration/singlecluster/webhook/core/localqueue_test.go
+++ b/test/integration/singlecluster/webhook/core/localqueue_test.go
@@ -17,10 +17,14 @@ limitations under the License.
 package core
 
 import (
+	"context"
+
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/test/util"
@@ -28,7 +32,15 @@ import (
 
 const queueName = "queue-test"
 
-var _ = ginkgo.Describe("Queue validating webhook", func() {
+var _ = ginkgo.Describe("Queue validating webhook", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
 	ginkgo.When("Updating a Queue", func() {
 		ginkgo.It("Should reject bad value for spec.clusterQueue", func() {
 			ginkgo.By("Creating a new Queue")
diff --git a/test/integration/singlecluster/webhook/core/resourceflavor_test.go b/test/integration/singlecluster/webhook/core/resourceflavor_test.go
index 9bfe02a66..068d30e20 100644
--- a/test/integration/singlecluster/webhook/core/resourceflavor_test.go
+++ b/test/integration/singlecluster/webhook/core/resourceflavor_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package core
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/onsi/ginkgo/v2"
@@ -25,7 +26,9 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/test/util"
@@ -36,7 +39,15 @@ const (
 	taintsMaxItems            = 8
 )
 
-var _ = ginkgo.Describe("ResourceFlavor Webhook", func() {
+var _ = ginkgo.Describe("ResourceFlavor Webhook", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
 	var ns *corev1.Namespace
 
 	ginkgo.BeforeEach(func() {
diff --git a/test/integration/singlecluster/webhook/core/suite_test.go b/test/integration/singlecluster/webhook/core/suite_test.go
index 4a2da3e37..d5dc40e2c 100644
--- a/test/integration/singlecluster/webhook/core/suite_test.go
+++ b/test/integration/singlecluster/webhook/core/suite_test.go
@@ -56,24 +56,25 @@ var _ = ginkgo.BeforeSuite(func() {
 	}
 	cfg = fwk.Init()
 	ctx, k8sClient = fwk.SetupClient(cfg)
-	fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
-		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
-		gomega.Expect(err).NotTo(gomega.HaveOccurred())
-
-		failedWebhook, err := webhooks.Setup(mgr)
-		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
-
-		cCache := cache.New(mgr.GetClient())
-		queues := queue.NewManager(mgr.GetClient(), cCache)
-
-		configuration := &config.Configuration{}
-		mgr.GetScheme().Default(configuration)
-
-		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
-		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
-	})
 })
 
 var _ = ginkgo.AfterSuite(func() {
 	fwk.Teardown()
 })
+
+func managerSetup(ctx context.Context, mgr manager.Manager, dispatcherName string) {
+	err := indexer.Setup(ctx, mgr.GetFieldIndexer())
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+	failedWebhook, err := webhooks.Setup(mgr, dispatcherName)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
+
+	cCache := cache.New(mgr.GetClient())
+	queues := queue.NewManager(mgr.GetClient(), cCache)
+
+	configuration := &config.Configuration{}
+	mgr.GetScheme().Default(configuration)
+
+	failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
+}
diff --git a/test/integration/singlecluster/webhook/core/workload_test.go b/test/integration/singlecluster/webhook/core/workload_test.go
index 80d941a88..8ec54636a 100644
--- a/test/integration/singlecluster/webhook/core/workload_test.go
+++ b/test/integration/singlecluster/webhook/core/workload_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package core
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -29,7 +30,9 @@ import (
 	"k8s.io/utils/clock"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -53,7 +56,16 @@ var _ = ginkgo.AfterEach(func() {
 	gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
 })
 
-var _ = ginkgo.Describe("Workload defaulting webhook", func() {
+var _ = ginkgo.Describe("Workload defaulting webhook", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+
 	ginkgo.Context("When creating a Workload", func() {
 		ginkgo.It("Should set default podSet name", func() {
 			ginkgo.By("Creating a new Workload")
@@ -100,7 +112,16 @@ var _ = ginkgo.Describe("Workload defaulting webhook", func() {
 	})
 })
 
-var _ = ginkgo.Describe("Workload validating webhook", func() {
+var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+
 	var realClock = clock.RealClock{}
 
 	ginkgo.Context("When creating a Workload", func() {
@@ -1031,3 +1052,235 @@ var _ = ginkgo.Describe("Workload validating webhook", func() {
 		})
 	})
 })
+
+var _ = ginkgo.Describe("Workload validating webhook ClusterName - Dispatcher AllAtOnce", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeAllAtOnce)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+	ginkgo.Context("When updating a Workload", func() {
+		var (
+			workloadPriorityClass *kueue.WorkloadPriorityClass
+			priorityClass         *schedulingv1.PriorityClass
+		)
+		ginkgo.BeforeEach(func() {
+			workloadPriorityClass = testing.MakeWorkloadPriorityClass("workload-priority-class").PriorityValue(200).Obj()
+			priorityClass = testing.MakePriorityClass("priority-class").PriorityValue(100).Obj()
+			util.MustCreate(ctx, k8sClient, workloadPriorityClass)
+			util.MustCreate(ctx, k8sClient, priorityClass)
+		})
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			gomega.Expect(k8sClient.Delete(ctx, workloadPriorityClass)).To(gomega.Succeed())
+			gomega.Expect(k8sClient.Delete(ctx, priorityClass)).To(gomega.Succeed())
+		})
+
+		ginkgo.DescribeTable("Validate Workload status on update",
+			func(setupWlStatus func(w *kueue.Workload), updateWlStatus func(w *kueue.Workload), setupMatcher, updateMatcher gomega.OmegaMatcher) {
+				ginkgo.By("Creating a new Workload")
+				workload := testing.MakeWorkloadWithGeneratedName(workloadName, ns.Name).Obj()
+				util.MustCreate(ctx, k8sClient, workload)
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					var wl kueue.Workload
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(workload), &wl)).To(gomega.Succeed())
+					setupWlStatus(&wl)
+					g.Expect(k8sClient.Status().Update(ctx, &wl)).Should(setupMatcher)
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					var wl kueue.Workload
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(workload), &wl)).To(gomega.Succeed())
+					updateWlStatus(&wl)
+					g.Expect(k8sClient.Status().Update(ctx, &wl)).Should(updateMatcher)
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			},
+			ginkgo.Entry("Valid: ClusterName is in NominatedClusters",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = []string{"worker1", "worker2"}
+					wl.Status.ClusterName = nil
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker2")
+				},
+				gomega.Succeed(),
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("Valid: ClusterName is not in NominatedClusters",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = []string{"worker1", "worker2"}
+					wl.Status.ClusterName = nil
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker3")
+				},
+				gomega.Succeed(),
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("Invalid: ClusterName is changed",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker1")
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker2")
+				},
+				gomega.Succeed(),
+				testing.BeInvalidError(),
+			),
+			ginkgo.Entry("Valid: ClusterName is set when NominatedClusters is empty",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = nil
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.ClusterName = ptr.To("worker1")
+					wl.Status.NominatedClusterNames = nil
+				},
+				gomega.Succeed(),
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("Invalid: ClusterName and NominatedClusters are mutually exclusive",
+				func(wl *kueue.Workload) {},
+				func(wl *kueue.Workload) {
+					wl.Status.ClusterName = ptr.To("worker1")
+					wl.Status.NominatedClusterNames = []string{"worker1", "worker2"}
+				},
+				gomega.Succeed(),
+				testing.BeInvalidError(),
+			),
+			ginkgo.Entry("Valid: neither ClusterName nor NominatedClusters is set",
+				func(wl *kueue.Workload) {},
+				func(wl *kueue.Workload) {
+					wl.Status.ClusterName = nil
+					wl.Status.NominatedClusterNames = nil
+				},
+				gomega.Succeed(),
+				gomega.Succeed(),
+			),
+		)
+	})
+})
+
+var _ = ginkgo.Describe("Workload validating webhook ClusterName - Dispatcher Incremental", ginkgo.Ordered, func() {
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, func(ctx context.Context, mgr manager.Manager) {
+			managerSetup(ctx, mgr, config.MultiKueueDispatcherModeIncremental)
+		})
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+	ginkgo.Context("When updating a Workload", func() {
+		var (
+			workloadPriorityClass *kueue.WorkloadPriorityClass
+			priorityClass         *schedulingv1.PriorityClass
+		)
+		ginkgo.BeforeEach(func() {
+			workloadPriorityClass = testing.MakeWorkloadPriorityClass("workload-priority-class").PriorityValue(200).Obj()
+			priorityClass = testing.MakePriorityClass("priority-class").PriorityValue(100).Obj()
+			util.MustCreate(ctx, k8sClient, workloadPriorityClass)
+			util.MustCreate(ctx, k8sClient, priorityClass)
+		})
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			gomega.Expect(k8sClient.Delete(ctx, workloadPriorityClass)).To(gomega.Succeed())
+			gomega.Expect(k8sClient.Delete(ctx, priorityClass)).To(gomega.Succeed())
+		})
+
+		ginkgo.DescribeTable("Validate Workload status on update",
+			func(setupWlStatus func(w *kueue.Workload), updateWlStatus func(w *kueue.Workload), setupMatcher, updateMatcher gomega.OmegaMatcher) {
+				ginkgo.By("Creating a new Workload")
+				workload := testing.MakeWorkloadWithGeneratedName(workloadName, ns.Name).Obj()
+				util.MustCreate(ctx, k8sClient, workload)
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					var wl kueue.Workload
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(workload), &wl)).To(gomega.Succeed())
+					setupWlStatus(&wl)
+					g.Expect(k8sClient.Status().Update(ctx, &wl)).Should(setupMatcher)
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					var wl kueue.Workload
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(workload), &wl)).To(gomega.Succeed())
+					updateWlStatus(&wl)
+					g.Expect(k8sClient.Status().Update(ctx, &wl)).Should(updateMatcher)
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			},
+			ginkgo.Entry("Valid: ClusterName is in NominatedClusters",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = []string{"worker1", "worker2"}
+					wl.Status.ClusterName = nil
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker2")
+				},
+				gomega.Succeed(),
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("Invalid: ClusterName is not in NominatedClusters",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = []string{"worker1", "worker2"}
+					wl.Status.ClusterName = nil
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker3")
+				},
+				gomega.Succeed(),
+				testing.BeForbiddenError(),
+			),
+			ginkgo.Entry("Invalid: ClusterName is changed",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker1")
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = ptr.To("worker2")
+				},
+				testing.BeForbiddenError(),
+				testing.BeForbiddenError(),
+			),
+			ginkgo.Entry("Invalid: ClusterName is set when NominatedClusters is empty",
+				func(wl *kueue.Workload) {
+					wl.Status.NominatedClusterNames = nil
+					wl.Status.ClusterName = nil
+				},
+				func(wl *kueue.Workload) {
+					wl.Status.ClusterName = ptr.To("worker1")
+					wl.Status.NominatedClusterNames = nil
+				},
+				gomega.Succeed(),
+				testing.BeForbiddenError(),
+			),
+			ginkgo.Entry("Invalid: ClusterName and NominatedClusters are mutually exclusive",
+				func(wl *kueue.Workload) {},
+				func(wl *kueue.Workload) {
+					wl.Status.ClusterName = ptr.To("worker1")
+					wl.Status.NominatedClusterNames = []string{"worker1", "worker2"}
+				},
+				gomega.Succeed(),
+				testing.BeInvalidError(),
+			),
+			ginkgo.Entry("Valid: neither ClusterName nor NominatedClusters is set",
+				func(wl *kueue.Workload) {},
+				func(wl *kueue.Workload) {
+					wl.Status.ClusterName = nil
+					wl.Status.NominatedClusterNames = nil
+				},
+				gomega.Succeed(),
+				gomega.Succeed(),
+			),
+		)
+	})
+})
diff --git a/test/integration/singlecluster/webhook/jobs/raycluster_webhook_test.go b/test/integration/singlecluster/webhook/jobs/raycluster_webhook_test.go
index 648bc24a6..be3bb7ac1 100644
--- a/test/integration/singlecluster/webhook/jobs/raycluster_webhook_test.go
+++ b/test/integration/singlecluster/webhook/jobs/raycluster_webhook_test.go
@@ -26,6 +26,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
@@ -93,7 +94,7 @@ var _ = ginkgo.Describe("RayCluster Webhook", func() {
 				gomega.Expect(err).NotTo(gomega.HaveOccurred())
 				jobframework.EnableIntegration(workloadrayjob.FrameworkName)
 
-				failedWebhook, err := webhooks.Setup(mgr)
+				failedWebhook, err := webhooks.Setup(mgr, config.MultiKueueDispatcherModeAllAtOnce)
 				gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
 				return nil