Repository: kubernetes-sigs/kueue

Issue:
MK Dispatcher: Extract Incremental mode to a separate controller
**What would you like to be added**:
Based on the [comment](https://github.com/kubernetes-sigs/kueue/pull/5782#discussion_r2201609035)
We would like to continue  to work on the MK Dispatcher and move Incremental mode implementation into separate controller.

**Why is this needed**:
The reference implementation would be proving that:
A. an external dispatcher can be written,
B. serving an example of how to write one
**Completion requirements**:

This enhancement requires the following artifacts:

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

The artifacts should be linked in subsequent comments.


Hints from issue comments:
@mimowo: Let me break it down into two separate issues:
- Move the incremental to a separate controller, but inside Kueue
- Create a separate blueprint project outside of Kueue binary implementing the dispatcher. For example adding clusters one-by-one

Let me retitle then
/retitle MK Dispatcher: Extract Incremental mode to a separate controller@mimowo: Split out the second point as: https://github.com/kubernetes-sigs/kueue/issues/6238@mszadkow: /assign@tenzen-y: Can we move the AllAtOnce dispatcher to a separate controller as well?
@mimowo: > Can we move the AllAtOnce dispatcher to a separate controller as well?

I don't see a blocker, might be a good idea to decompose the code more.@tenzen-y: > > Can we move the AllAtOnce dispatcher to a separate controller as well?
> 
> I don't see a blocker, might be a good idea to decompose the code more.

I would still like to decouple the AllAtOnce dispatcher from AC MultiKueue Workload controller to a dedicated reconciler within the kueue-controller-manager the same as incremental dispatcher.
@mimowo: sgtm, would you like to open a follow up issue?@mimowo: Actually, I just opened: https://github.com/kubernetes-sigs/kueue/issues/6803@tenzen-y: > Actually, I just opened: [#6803](https://github.com/kubernetes-sigs/kueue/issues/6803)

Thank you!

Solution patch:
diff --git a/cmd/kueue/main.go b/cmd/kueue/main.go
index 7768a3687..89f359fa7 100644
--- a/cmd/kueue/main.go
+++ b/cmd/kueue/main.go
@@ -62,6 +62,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/tas"
 	tasindexer "sigs.k8s.io/kueue/pkg/controller/tas/indexer"
+	dispatcher "sigs.k8s.io/kueue/pkg/controller/workloaddispatcher"
 	"sigs.k8s.io/kueue/pkg/debugger"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/metrics"
@@ -350,6 +351,10 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *schdcache.C
 		); err != nil {
 			return fmt.Errorf("could not setup MultiKueue controller: %w", err)
 		}
+
+		if failedDispatcher, err := dispatcher.SetupControllers(mgr, cfg, ptr.Deref(cfg.MultiKueue.DispatcherName, configapi.MultiKueueDispatcherModeAllAtOnce)); err != nil {
+			return fmt.Errorf("could not setup Dispatcher controller %q for MultiKueue: %w", failedDispatcher, err)
+		}
 	}
 
 	if features.Enabled(features.TopologyAwareScheduling) {
diff --git a/pkg/controller/admissionchecks/multikueue/admissioncheck.go b/pkg/controller/admissionchecks/multikueue/admissioncheck.go
index 048ad9707..643857967 100644
--- a/pkg/controller/admissionchecks/multikueue/admissioncheck.go
+++ b/pkg/controller/admissionchecks/multikueue/admissioncheck.go
@@ -37,18 +37,12 @@ import (
 	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
 )
 
-type multiKueueStoreHelper = admissioncheck.ConfigHelper[*kueue.MultiKueueConfig, kueue.MultiKueueConfig]
-
-func newMultiKueueStoreHelper(c client.Client) (*multiKueueStoreHelper, error) {
-	return admissioncheck.NewConfigHelper[*kueue.MultiKueueConfig](c)
-}
-
 // ACReconciler implements the reconciler for all the admission checks controlled by multikueue.
 // Its main task being to maintain the active state of the admission checks based on the heath
 // of its referenced MultiKueueClusters.
 type ACReconciler struct {
 	client client.Client
-	helper *multiKueueStoreHelper
+	helper *admissioncheck.MultiKueueStoreHelper
 }
 
 var _ reconcile.Reconciler = (*ACReconciler)(nil)
@@ -137,7 +131,7 @@ func (a *ACReconciler) Reconcile(ctx context.Context, req reconcile.Request) (re
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=admissionchecks,verbs=get;list;watch
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=multikueueconfigs,verbs=get;list;watch
 
-func newACReconciler(c client.Client, helper *multiKueueStoreHelper) *ACReconciler {
+func newACReconciler(c client.Client, helper *admissioncheck.MultiKueueStoreHelper) *ACReconciler {
 	return &ACReconciler{
 		client: c,
 		helper: helper,
diff --git a/pkg/controller/admissionchecks/multikueue/controllers.go b/pkg/controller/admissionchecks/multikueue/controllers.go
index 9be4c62b6..888b11f85 100644
--- a/pkg/controller/admissionchecks/multikueue/controllers.go
+++ b/pkg/controller/admissionchecks/multikueue/controllers.go
@@ -24,6 +24,7 @@ import (
 	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
 )
 
 const (
@@ -103,7 +104,7 @@ func SetupControllers(mgr ctrl.Manager, namespace string, opts ...SetupOption) e
 		o(options)
 	}
 
-	helper, err := newMultiKueueStoreHelper(mgr.GetClient())
+	helper, err := admissioncheck.NewMultiKueueStoreHelper(mgr.GetClient())
 	if err != nil {
 		return err
 	}
diff --git a/pkg/controller/admissionchecks/multikueue/workload.go b/pkg/controller/admissionchecks/multikueue/workload.go
index d806ea040..989523569 100644
--- a/pkg/controller/admissionchecks/multikueue/workload.go
+++ b/pkg/controller/admissionchecks/multikueue/workload.go
@@ -23,7 +23,6 @@ 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"
@@ -31,7 +30,6 @@ import (
 	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"
@@ -53,18 +51,13 @@ import (
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
-const (
-	incrementalDispatcherRoundTimeout = 5 * time.Minute
-)
-
 var (
-	realClock           = clock.RealClock{}
-	errNoActiveClusters = errors.New("no active clusters")
+	realClock = clock.RealClock{}
 )
 
 type wlReconciler struct {
 	client            client.Client
-	helper            *multiKueueStoreHelper
+	helper            *admissioncheck.MultiKueueStoreHelper
 	clusters          *clustersReconciler
 	origin            string
 	workerLostTimeout time.Duration
@@ -74,7 +67,6 @@ type wlReconciler struct {
 	recorder          record.EventRecorder
 	clock             clock.Clock
 	dispatcherName    string
-	roundStartTimes   map[types.NamespacedName]time.Time
 }
 
 var _ reconcile.Reconciler = (*wlReconciler)(nil)
@@ -181,7 +173,7 @@ func (w *wlReconciler) Reconcile(ctx context.Context, req reconcile.Request) (re
 		isDeleted = !wl.DeletionTimestamp.IsZero()
 	}
 
-	mkAc, err := w.multikueueAC(ctx, wl)
+	mkAc, err := admissioncheck.GetMultiKueueAdmissionCheck(ctx, w.client, wl)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
@@ -268,23 +260,11 @@ func (w *wlReconciler) remoteClientsForAC(ctx context.Context, acName kueue.Admi
 		}
 	}
 	if len(clients) == 0 {
-		return nil, errNoActiveClusters
+		return nil, admissioncheck.ErrNoActiveClusters
 	}
 	return clients, nil
 }
 
-func (w *wlReconciler) multikueueAC(ctx context.Context, local *kueue.Workload) (*kueue.AdmissionCheckState, error) {
-	relevantChecks, err := admissioncheck.FilterForController(ctx, w.client, local.Status.AdmissionChecks, kueue.MultiKueueControllerName)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(relevantChecks) == 0 {
-		return nil, nil
-	}
-	return admissioncheck.FindAdmissionCheck(local.Status.AdmissionChecks, relevantChecks[0]), nil
-}
-
 func (w *wlReconciler) adapter(local *kueue.Workload) (jobframework.MultiKueueAdapter, *metav1.OwnerReference) {
 	if controller := metav1.GetControllerOf(local); controller != nil {
 		adapterKey := schema.FromAPIVersionAndKind(controller.APIVersion, controller.Kind).String()
@@ -449,10 +429,8 @@ func (w *wlReconciler) nominateAndSynchronizeWorkers(ctx context.Context, group
 	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:
+	if w.dispatcherName == config.MultiKueueDispatcherModeAllAtOnce {
 		for workerName := range group.remotes {
 			nominatedWorkers = append(nominatedWorkers, workerName)
 		}
@@ -463,36 +441,11 @@ func (w *wlReconciler) nominateAndSynchronizeWorkers(ctx context.Context, group
 				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)
+	} else {
+		// Incremental dispatcher and External dispatcher path
 		nominatedWorkers = group.local.Status.NominatedClusterNames
 	}
-
-	log.V(4).Info("Synchronize nominated worker clusters", "nominatedWorkerClusterNames", nominatedWorkers)
+	log.V(4).Info("Synchronize nominated worker clusters", "dispatcherName", w.dispatcherName, "nominatedWorkerClusterNames", nominatedWorkers)
 
 	var errs []error
 	for rem, remoteWl := range group.remotes {
@@ -512,32 +465,7 @@ func (w *wlReconciler) nominateAndSynchronizeWorkers(ctx context.Context, group
 			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)
-		}
-	}
-	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
+	return reconcile.Result{}, errors.Join(errs...)
 }
 
 func (w *wlReconciler) Create(_ event.CreateEvent) bool {
@@ -559,7 +487,7 @@ func (w *wlReconciler) Generic(_ event.GenericEvent) bool {
 	return true
 }
 
-func newWlReconciler(c client.Client, helper *multiKueueStoreHelper, cRec *clustersReconciler, origin string,
+func newWlReconciler(c client.Client, helper *admissioncheck.MultiKueueStoreHelper, cRec *clustersReconciler, origin string,
 	recorder record.EventRecorder, workerLostTimeout, eventsBatchPeriod time.Duration,
 	adapters map[string]jobframework.MultiKueueAdapter, dispatcherName string,
 	options ...Option,
@@ -576,7 +504,6 @@ func newWlReconciler(c client.Client, helper *multiKueueStoreHelper, cRec *clust
 		recorder:          recorder,
 		clock:             realClock,
 		dispatcherName:    dispatcherName,
-		roundStartTimes:   make(map[types.NamespacedName]time.Time),
 	}
 	for _, option := range options {
 		option(r)
diff --git a/pkg/controller/workloaddispatcher/controllers.go b/pkg/controller/workloaddispatcher/controllers.go
new file mode 100644
index 000000000..59bb0da5e
--- /dev/null
+++ b/pkg/controller/workloaddispatcher/controllers.go
@@ -0,0 +1,39 @@
+/*
+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 dispatcher
+
+import (
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
+	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
+)
+
+func SetupControllers(mgr ctrl.Manager, cfg *configapi.Configuration, dispatcherName string) (string, error) {
+	helper, err := admissioncheck.NewMultiKueueStoreHelper(mgr.GetClient())
+	if err != nil {
+		return "", err
+	}
+
+	idRec := NewIncrementalDispatcherReconciler(mgr.GetClient(), helper, dispatcherName)
+	err = idRec.SetupWithManager(mgr, cfg)
+	if err != nil {
+		return "multikueue-incremental-dispatcher", err
+	}
+
+	return "", nil
+}
diff --git a/pkg/controller/workloaddispatcher/incrementaldispatcher.go b/pkg/controller/workloaddispatcher/incrementaldispatcher.go
new file mode 100644
index 000000000..1618930bc
--- /dev/null
+++ b/pkg/controller/workloaddispatcher/incrementaldispatcher.go
@@ -0,0 +1,199 @@
+/*
+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 dispatcher
+
+import (
+	"context"
+	"errors"
+	"slices"
+	"time"
+
+	"github.com/go-logr/logr"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/utils/clock"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	kueueconfig "sigs.k8s.io/kueue/apis/config/v1beta1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/controller/core"
+	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
+	utilmaps "sigs.k8s.io/kueue/pkg/util/maps"
+	"sigs.k8s.io/kueue/pkg/workload"
+)
+
+const (
+	incrementalDispatcherRoundTimeout = 5 * time.Minute
+)
+
+var ErrNoMoreWorkers = errors.New("no more workers to nominate")
+
+type IncrementalDispatcherReconciler struct {
+	client          client.Client
+	helper          *admissioncheck.MultiKueueStoreHelper
+	clock           clock.Clock
+	dispatcherName  string
+	roundStartTimes *utilmaps.SyncMap[types.NamespacedName, time.Time]
+}
+
+var realClock = clock.RealClock{}
+var _ reconcile.Reconciler = (*IncrementalDispatcherReconciler)(nil)
+
+func (r *IncrementalDispatcherReconciler) SetupWithManager(mgr ctrl.Manager, cfg *kueueconfig.Configuration) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		Named("multikueue_incremental_dispatcher").
+		For(&kueue.Workload{}).
+		Complete(core.WithLeadingManager(mgr, r, &kueue.Workload{}, cfg))
+}
+
+func NewIncrementalDispatcherReconciler(c client.Client, helper *admissioncheck.MultiKueueStoreHelper, dispatcherName string) *IncrementalDispatcherReconciler {
+	return &IncrementalDispatcherReconciler{
+		client:          c,
+		helper:          helper,
+		clock:           realClock,
+		dispatcherName:  dispatcherName,
+		roundStartTimes: utilmaps.NewSyncMap[types.NamespacedName, time.Time](0),
+	}
+}
+func (r *IncrementalDispatcherReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := ctrl.LoggerFrom(ctx)
+	log.V(3).Info("Nominate Worker Clusters with Incremental Dispatcher")
+
+	if r.dispatcherName != kueueconfig.MultiKueueDispatcherModeIncremental {
+		log.V(3).Info("Not a Incremental Dispatcher, skip the reconciliation", "dispatcherName", r.dispatcherName)
+		return reconcile.Result{}, nil
+	}
+
+	wl := &kueue.Workload{}
+	err := r.client.Get(ctx, req.NamespacedName, wl)
+	if err != nil {
+		log.Error(err, "Failed to retrieve Workload, skip the reconciliation")
+		if apierrors.IsNotFound(err) {
+			r.clearRoundStartTime(req.NamespacedName)
+		}
+		return reconcile.Result{}, err
+	}
+
+	if !wl.DeletionTimestamp.IsZero() {
+		log.V(3).Info("Workload is deleted, skip the reconciliation")
+		r.clearRoundStartTime(req.NamespacedName)
+		return reconcile.Result{}, nil
+	}
+
+	mkAc, err := admissioncheck.GetMultiKueueAdmissionCheck(ctx, r.client, wl)
+	if err != nil {
+		log.Error(err, "Can not get MultiKueue AdmissionCheckState")
+		return reconcile.Result{}, err
+	}
+
+	if mkAc == nil || mkAc.State != kueue.CheckStatePending {
+		log.V(3).Info("AdmissionCheckState is not in Pending, skip the reconciliation")
+		return reconcile.Result{}, nil
+	}
+
+	// The workload is already assigned to a cluster, no need to nominate workers.
+	if wl.Status.ClusterName != nil {
+		log.V(3).Info("The workload is already assigned to a cluster, no need to nominate workers")
+		return reconcile.Result{}, nil
+	}
+
+	remoteClusters, err := admissioncheck.GetRemoteClusters(ctx, r.helper, mkAc.Name)
+	if err != nil {
+		log.Error(err, "Can not get workload group")
+		return reconcile.Result{}, err
+	}
+
+	if workload.IsFinished(wl) || !workload.HasQuotaReservation(wl) {
+		log.V(3).Info("Workload is already finished or has no quota reserved, skip the reconciliation")
+		r.clearRoundStartTime(req.NamespacedName)
+		return reconcile.Result{}, nil
+	}
+
+	return r.nominateWorkers(ctx, wl, remoteClusters, log)
+}
+
+func (r *IncrementalDispatcherReconciler) nominateWorkers(ctx context.Context, wl *kueue.Workload, remoteClusters sets.Set[string], log logr.Logger) (reconcile.Result, error) {
+	key := client.ObjectKeyFromObject(wl)
+	roundStart, found := r.getRoundStartTime(key)
+	now := r.clock.Now()
+	log.V(5).Info("nominating worker clusters", "nominatedAt", roundStart, "revokedAt", roundStart.Add(incrementalDispatcherRoundTimeout))
+	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, wl, remoteClusters)
+	log.V(5).Info("revoke outdated nomination and nominate new worker clusters", "revokedWorkerClusters", wl.Status.NominatedClusterNames, "nominatedWorkerClusters", nextNominatedWorkers)
+	if err != nil {
+		log.Error(err, "Failed to nominate next worker clusters")
+		return reconcile.Result{}, err
+	}
+
+	nominatedWorkers := append(wl.Status.NominatedClusterNames, nextNominatedWorkers...)
+	wl.Status.NominatedClusterNames = nominatedWorkers
+	if err := workload.ApplyAdmissionStatus(ctx, r.client, wl, true, r.clock); err != nil {
+		log.V(2).Error(err, "Failed to patch nominated clusters")
+		return reconcile.Result{}, err
+	}
+	// only update the round start time if we successfully nominated workers
+	r.setRoundStartTime(key, now)
+
+	return reconcile.Result{}, nil
+}
+
+// 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, wl *kueue.Workload, remoteClusters sets.Set[string]) ([]string, error) {
+	alreadyNominated := sets.New(wl.Status.NominatedClusterNames...)
+
+	workers := make([]string, 0, len(remoteClusters))
+	for remoteWorker := range remoteClusters {
+		if !alreadyNominated.Has(remoteWorker) {
+			workers = append(workers, remoteWorker)
+		}
+	}
+	slices.Sort(workers)
+
+	log.V(5).Info("proceeding worker clusters nomination", "alreadyNominatedClusterNames", alreadyNominated, "remainingClusterNames", workers)
+
+	if len(workers) == 0 {
+		return nil, ErrNoMoreWorkers
+	}
+	batchSize := 3
+	if len(workers) < batchSize {
+		return workers, nil
+	}
+	return workers[:batchSize], nil
+}
+
+func (r *IncrementalDispatcherReconciler) setRoundStartTime(key types.NamespacedName, t time.Time) {
+	r.roundStartTimes.Add(key, t)
+}
+
+func (r *IncrementalDispatcherReconciler) getRoundStartTime(key types.NamespacedName) (time.Time, bool) {
+	t, found := r.roundStartTimes.Get(key)
+	return t, found
+}
+
+// clearRoundStartTime removes the round start time for the given workload key.
+func (r *IncrementalDispatcherReconciler) clearRoundStartTime(key types.NamespacedName) {
+	r.roundStartTimes.Delete(key)
+}
diff --git a/pkg/util/admissioncheck/admissioncheck.go b/pkg/util/admissioncheck/admissioncheck.go
index ea0d05fb4..0b0454ecb 100644
--- a/pkg/util/admissioncheck/admissioncheck.go
+++ b/pkg/util/admissioncheck/admissioncheck.go
@@ -35,6 +35,7 @@ import (
 var (
 	ErrNilParametersRef = errors.New("missing parameters reference")
 	ErrBadParametersRef = errors.New("bad parameters reference")
+	ErrNoActiveClusters = errors.New("no active clusters")
 )
 
 type objAsPtr[T any] interface {
@@ -47,6 +48,12 @@ type ConfigHelper[PtrT objAsPtr[T], T any] struct {
 	client client.Client
 }
 
+type MultiKueueStoreHelper = ConfigHelper[*kueue.MultiKueueConfig, kueue.MultiKueueConfig]
+
+func NewMultiKueueStoreHelper(c client.Client) (*MultiKueueStoreHelper, error) {
+	return NewConfigHelper[*kueue.MultiKueueConfig](c)
+}
+
 func NewConfigHelper[PtrT objAsPtr[T], T any](c client.Client) (*ConfigHelper[PtrT, T], error) {
 	helper := ConfigHelper[PtrT, T]{
 		client: c,
@@ -183,3 +190,33 @@ func FindAdmissionCheck(checks []kueue.AdmissionCheckState, checkName kueue.Admi
 
 	return nil
 }
+
+func GetMultiKueueAdmissionCheck(ctx context.Context, c client.Client, wl *kueue.Workload) (*kueue.AdmissionCheckState, error) {
+	relevantChecks, err := FilterForController(ctx, c, wl.Status.AdmissionChecks, kueue.MultiKueueControllerName)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(relevantChecks) == 0 {
+		return nil, nil
+	}
+
+	for _, check := range wl.Status.AdmissionChecks {
+		if check.Name == relevantChecks[0] {
+			return &check, nil
+		}
+	}
+	return nil, nil
+}
+
+func GetRemoteClusters(ctx context.Context, helper *MultiKueueStoreHelper, acName kueue.AdmissionCheckReference) (sets.Set[string], error) {
+	cfg, err := helper.ConfigForAdmissionCheck(ctx, acName)
+	if err != nil {
+		return nil, err
+	}
+	if len(cfg.Spec.Clusters) == 0 {
+		return nil, ErrNoActiveClusters
+	}
+
+	return sets.New(cfg.Spec.Clusters...), nil
+}



Test patch:
diff --git a/pkg/controller/admissionchecks/multikueue/admissioncheck_test.go b/pkg/controller/admissionchecks/multikueue/admissioncheck_test.go
index 52b5b85b9..3935ea857 100644
--- a/pkg/controller/admissionchecks/multikueue/admissioncheck_test.go
+++ b/pkg/controller/admissionchecks/multikueue/admissioncheck_test.go
@@ -26,6 +26,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 )
 
@@ -253,7 +254,7 @@ func TestReconcile(t *testing.T) {
 
 			c := builder.Build()
 
-			helper, _ := newMultiKueueStoreHelper(c)
+			helper, _ := admissioncheck.NewMultiKueueStoreHelper(c)
 			reconciler := newACReconciler(c, helper)
 
 			_, gotErr := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: tc.reconcileFor}})
diff --git a/pkg/controller/admissionchecks/multikueue/workload_test.go b/pkg/controller/admissionchecks/multikueue/workload_test.go
index 5943633d3..695a349bb 100644
--- a/pkg/controller/admissionchecks/multikueue/workload_test.go
+++ b/pkg/controller/admissionchecks/multikueue/workload_test.go
@@ -44,6 +44,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/features"
+	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
 	"sigs.k8s.io/kueue/pkg/util/slices"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
@@ -337,7 +338,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"}).
+					NominatedClusterNames("worker1", "worker2").
 					Obj(),
 			},
 			wantWorker1Workloads: []kueue.Workload{
@@ -370,7 +371,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"}).
+					NominatedClusterNames("worker1", "worker2").
 					Obj(),
 			},
 			wantWorker1Workloads: []kueue.Workload{
@@ -1193,7 +1194,7 @@ func TestWlReconcile(t *testing.T) {
 				cRec.remoteClients["worker2"] = w2remoteClient
 			}
 
-			helper, _ := newMultiKueueStoreHelper(managerClient)
+			helper, _ := admissioncheck.NewMultiKueueStoreHelper(managerClient)
 			recorder := &utiltesting.EventRecorder{}
 			mkDispatcherName := ptr.Deref(tc.dispatcherName, config.MultiKueueDispatcherModeAllAtOnce)
 			reconciler := newWlReconciler(managerClient, helper, cRec, defaultOrigin, recorder, defaultWorkerLostTimeout, time.Second, adapters, mkDispatcherName, WithClock(t, fakeClock))
@@ -1310,71 +1311,25 @@ func TestNominateAndSynchronizeWorkers_MoreCases(t *testing.T) {
 		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,
 		},
+		// Incremental dispatcher tests were moved to a separate file.
 		{
 			name:           "External controller: no nominated workers, nothing created",
 			dispatcherMode: externalMultiKueueDispatcherController,
 			remotes:        remotes,
 			wantCreated:    nil,
-			wantRetryAfter: 0,
 		},
 		{
 			name:             "External controller: nominate remote1 and remote6",
@@ -1446,20 +1401,13 @@ func TestNominateAndSynchronizeWorkers_MoreCases(t *testing.T) {
 			}
 
 			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))
+				clock:          fakeClock,
+				dispatcherName: tt.dispatcherMode,
+				client:         wlClientBuilder.Build(),
 			}
 
 			ctx, _ := utiltesting.ContextWithLog(t)
-			res, err := wlRec.nominateAndSynchronizeWorkers(ctx, group)
+			_, err := wlRec.nominateAndSynchronizeWorkers(ctx, group)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("expected error: %v, got: %v", tt.wantErr, err)
 			}
@@ -1475,17 +1423,6 @@ func TestNominateAndSynchronizeWorkers_MoreCases(t *testing.T) {
 			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/workloaddispatcher/incrementaldispatcher_test.go b/pkg/controller/workloaddispatcher/incrementaldispatcher_test.go
new file mode 100644
index 000000000..703a75fb8
--- /dev/null
+++ b/pkg/controller/workloaddispatcher/incrementaldispatcher_test.go
@@ -0,0 +1,337 @@
+/*
+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 dispatcher
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	testingclock "k8s.io/utils/clock/testing"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
+	kueueconfig "sigs.k8s.io/kueue/apis/config/v1beta1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
+	utilmaps "sigs.k8s.io/kueue/pkg/util/maps"
+	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+)
+
+func TestIncrementalDispatcherReconciler_Reconcile(t *testing.T) {
+	const workloadName = "test-workload"
+
+	now := time.Now()
+	fakeClock := testingclock.NewFakeClock(now)
+	baseWorkload := utiltesting.MakeWorkload(workloadName, metav1.NamespaceDefault)
+
+	tests := map[string]struct {
+		dispatcherName string
+		workload       *kueue.Workload
+		mkAcState      *kueue.AdmissionCheckState
+		wantErr        error
+		remoteClusters []string
+		clusters       []kueue.MultiKueueCluster
+	}{
+		"dispatcher name mismatch": {
+			dispatcherName: "other",
+			workload:       baseWorkload.Clone().Obj(),
+		},
+		"workload not found": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       nil,
+			wantErr:        apierrors.NewNotFound(schema.GroupResource{Group: kueue.GroupVersion.Group, Resource: "workloads"}, workloadName),
+		},
+		"workload deleted": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       baseWorkload.Clone().DeletionTimestamp(now).Finalizers("kubernetes").Obj(),
+		},
+		"admission check nil": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       baseWorkload.Clone().Obj(),
+		},
+		"admission check is rejected": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       baseWorkload.Clone().Obj(),
+			mkAcState: &kueue.AdmissionCheckState{
+				Name:  "ac1",
+				State: kueue.CheckStateRejected,
+			},
+		},
+		"admission check is ready": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       baseWorkload.Clone().Obj(),
+			mkAcState: &kueue.AdmissionCheckState{
+				Name:  "ac1",
+				State: kueue.CheckStateReady,
+			},
+		},
+		"already assigned to cluster": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       baseWorkload.Clone().ClusterName("assigned").Obj(),
+			mkAcState: &kueue.AdmissionCheckState{
+				Name:  "ac1",
+				State: kueue.CheckStatePending,
+			},
+		},
+		"workload is already finished": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       baseWorkload.Clone().Finished().Obj(),
+			mkAcState: &kueue.AdmissionCheckState{
+				Name:  "ac1",
+				State: kueue.CheckStatePending,
+			},
+			remoteClusters: []string{"cluster1"},
+			clusters: []kueue.MultiKueueCluster{
+				*utiltesting.MakeMultiKueueCluster("cluster1").
+					KubeConfig(kueue.SecretLocationType, "cluster1").
+					Generation(1).
+					Obj(),
+			},
+		},
+		"workload has quota reserved": {
+			dispatcherName: kueueconfig.MultiKueueDispatcherModeIncremental,
+			workload:       baseWorkload.Clone().Obj(),
+			mkAcState: &kueue.AdmissionCheckState{
+				Name:  "ac1",
+				State: kueue.CheckStatePending,
+			},
+			remoteClusters: []string{"cluster1"},
+			clusters: []kueue.MultiKueueCluster{
+				*utiltesting.MakeMultiKueueCluster("cluster1").
+					KubeConfig(kueue.SecretLocationType, "cluster1").
+					Generation(1).
+					Obj(),
+			},
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			objs := []client.Object{}
+			if tc.mkAcState != nil {
+				tc.workload.Status.AdmissionChecks = []kueue.AdmissionCheckState{*tc.mkAcState}
+				ac := utiltesting.MakeAdmissionCheck(string(tc.mkAcState.Name)).
+					ControllerName(kueue.MultiKueueControllerName).
+					Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", string(tc.mkAcState.Name)).
+					Obj()
+
+				objs = append(objs, ac)
+			}
+
+			if tc.workload != nil {
+				objs = append(objs, tc.workload)
+			}
+
+			if tc.mkAcState != nil {
+				mkConfig := utiltesting.MakeMultiKueueConfig(string(tc.mkAcState.Name)).Clusters("cluster1").Obj()
+				objs = append(objs, mkConfig)
+			}
+			scheme := runtime.NewScheme()
+			if err := kueue.AddToScheme(scheme); err != nil {
+				t.Fatalf("Fail to add to scheme %s", err)
+			}
+
+			if tc.clusters != nil {
+				for _, cluster := range tc.clusters {
+					objs = append(objs, &cluster)
+				}
+			}
+			cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build()
+			helper, _ := admissioncheck.NewMultiKueueStoreHelper(cl)
+			rec := &IncrementalDispatcherReconciler{
+				client:          cl,
+				helper:          helper,
+				clock:           fakeClock,
+				dispatcherName:  tc.dispatcherName,
+				roundStartTimes: utilmaps.NewSyncMap[types.NamespacedName, time.Time](0),
+			}
+
+			req := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: metav1.NamespaceDefault, Name: workloadName}}
+			ctx, _ := utiltesting.ContextWithLog(t)
+			_, gotErr := rec.Reconcile(ctx, req)
+			if diff := cmp.Diff(tc.wantErr, gotErr); diff != "" {
+				t.Errorf("Unexpected error (-want/+got)\n%s", diff)
+			}
+		})
+	}
+}
+
+func TestIncrementalDispatcherNominateWorkers(t *testing.T) {
+	const testName = "test-wl"
+	now := time.Now()
+	fakeClock := testingclock.NewFakeClock(now)
+	baseWl := utiltesting.MakeWorkload(testName, metav1.NamespaceDefault).
+		AdmissionCheck(kueue.AdmissionCheckState{
+			Name:  "ac1",
+			State: kueue.CheckStatePending,
+		})
+
+	testCases := map[string]struct {
+		remoteClusters             sets.Set[string]
+		workload                   *kueue.Workload
+		wantNominatedClustersCount int
+		wantErr                    error
+		advanceRoundTime           bool
+		wantNominatedClusters      []string
+	}{
+		"one remote": {
+			remoteClusters:             sets.New("A"),
+			workload:                   baseWl.Clone().Obj(),
+			wantNominatedClustersCount: 1,
+			wantErr:                    nil,
+			advanceRoundTime:           false,
+			wantNominatedClusters:      []string{"A"},
+		},
+		"two remotes": {
+			remoteClusters:             sets.New("A", "B"),
+			workload:                   baseWl.Clone().Obj(),
+			wantNominatedClustersCount: 2,
+			wantErr:                    nil,
+			advanceRoundTime:           false,
+			wantNominatedClusters:      []string{"A", "B"},
+		},
+		"three remotes": {
+			remoteClusters:             sets.New("A", "B", "C"),
+			workload:                   baseWl.Clone().Obj(),
+			wantNominatedClustersCount: 3,
+			wantErr:                    nil,
+			advanceRoundTime:           false,
+			wantNominatedClusters:      []string{"A", "B", "C"},
+		},
+		"fifteen remotes": {
+			remoteClusters:             sets.New("A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L", "M", "N", "O"),
+			workload:                   baseWl.Clone().Obj(),
+			wantNominatedClustersCount: 3,
+			wantErr:                    nil,
+			advanceRoundTime:           false,
+			wantNominatedClusters:      []string{"A", "B", "C"},
+		},
+		"all already nominated (3)": {
+			remoteClusters:             sets.New("A", "B", "C"),
+			workload:                   baseWl.Clone().NominatedClusterNames("A", "B", "C").Obj(),
+			wantNominatedClustersCount: 3,
+			wantErr:                    ErrNoMoreWorkers,
+			advanceRoundTime:           true,
+			wantNominatedClusters:      []string{"A", "B", "C"},
+		},
+		"all already nominated (1)": {
+			remoteClusters:             sets.New("A"),
+			workload:                   baseWl.Clone().NominatedClusterNames("A").Obj(),
+			wantNominatedClustersCount: 1,
+			wantErr:                    ErrNoMoreWorkers,
+			advanceRoundTime:           true,
+			wantNominatedClusters:      []string{"A"},
+		},
+		"round expired, next set nominated": {
+			remoteClusters:             sets.New("A", "B", "C", "D", "E", "F"),
+			workload:                   baseWl.Clone().NominatedClusterNames("A", "B", "C").Obj(),
+			wantNominatedClustersCount: 6,
+			wantErr:                    nil,
+			advanceRoundTime:           true,
+			wantNominatedClusters:      []string{"A", "B", "C", "D", "E", "F"},
+		},
+		"round in progress, keep current": {
+			remoteClusters:             sets.New("A", "B", "C", "D", "E", "F"),
+			workload:                   baseWl.Clone().NominatedClusterNames("A", "B", "C").Obj(),
+			wantNominatedClustersCount: 3,
+			wantErr:                    nil,
+			advanceRoundTime:           false,
+			wantNominatedClusters:      []string{"A", "B", "C"},
+		},
+		"round expired, nominate all": {
+			remoteClusters:             sets.New("A", "B", "C", "D", "E", "F", "G", "H"),
+			workload:                   baseWl.Clone().NominatedClusterNames("A", "B", "C", "D", "E", "F").Obj(),
+			wantNominatedClustersCount: 8,
+			wantErr:                    nil,
+			advanceRoundTime:           true,
+			wantNominatedClusters:      []string{"A", "B", "C", "D", "E", "F", "G", "H"},
+		},
+		"no remotes": {
+			remoteClusters:             make(sets.Set[string]),
+			workload:                   baseWl.Clone().Obj(),
+			wantNominatedClustersCount: 0,
+			wantErr:                    ErrNoMoreWorkers,
+			advanceRoundTime:           false,
+			wantNominatedClusters:      []string{},
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			scheme := runtime.NewScheme()
+			if err := kueue.AddToScheme(scheme); err != nil {
+				t.Fatalf("Fail to add to scheme %s", err)
+			}
+
+			objs := []client.Object{tc.workload}
+			client := fake.NewClientBuilder().WithScheme(scheme).
+				WithInterceptorFuncs(interceptor.Funcs{
+					SubResourcePatch: func(ctx context.Context, client client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+						tc.workload.Status.NominatedClusterNames = obj.(*kueue.Workload).Status.NominatedClusterNames
+						return utiltesting.TreatSSAAsStrategicMerge(ctx, client, subResourceName, obj, patch, opts...)
+					},
+				}).WithObjects(objs...).WithStatusSubresource(objs...).Build()
+
+			reconciler := &IncrementalDispatcherReconciler{
+				client:          client,
+				clock:           fakeClock,
+				roundStartTimes: utilmaps.NewSyncMap[types.NamespacedName, time.Time](0),
+			}
+
+			key := types.NamespacedName{Namespace: tc.workload.Namespace, Name: tc.workload.Name}
+			if tc.advanceRoundTime {
+				reconciler.setRoundStartTime(key, fakeClock.Now().Add(-incrementalDispatcherRoundTimeout-time.Second))
+			} else if tc.workload.Status.NominatedClusterNames != nil {
+				reconciler.setRoundStartTime(key, fakeClock.Now())
+			}
+
+			previousRoundNominatedClusters := tc.workload.Status.NominatedClusterNames
+
+			ctx, log := utiltesting.ContextWithLog(t)
+			_, gotErr := reconciler.nominateWorkers(ctx, tc.workload, tc.remoteClusters, log)
+
+			if diff := cmp.Diff(tc.wantErr, gotErr, cmpopts.EquateErrors()); diff != "" {
+				t.Errorf("Unexpected error (-want/+got)\n%s", diff)
+			}
+
+			if tc.wantErr != nil && tc.workload.Status.NominatedClusterNames != nil {
+				if len(tc.workload.Status.NominatedClusterNames) != len(previousRoundNominatedClusters) {
+					t.Errorf("expected nominated clusters to remain unchanged, got %v", tc.workload.Status.NominatedClusterNames)
+				}
+			}
+
+			if tc.advanceRoundTime && tc.wantNominatedClusters != nil {
+				if diff := cmp.Diff(tc.wantNominatedClusters, tc.workload.Status.NominatedClusterNames); diff != "" {
+					t.Errorf("unexpected nominated clusters (-want/+got):\n%s", diff)
+				}
+			} else if len(tc.workload.Status.NominatedClusterNames) != tc.wantNominatedClustersCount {
+				t.Errorf("expected %d nominated clusters, got %d: %v", tc.wantNominatedClustersCount, len(tc.workload.Status.NominatedClusterNames), tc.workload.Status.NominatedClusterNames)
+			}
+		})
+	}
+}
diff --git a/pkg/util/admissioncheck/admissioncheck_test.go b/pkg/util/admissioncheck/admissioncheck_test.go
index 8a588ac48..d6c872409 100644
--- a/pkg/util/admissioncheck/admissioncheck_test.go
+++ b/pkg/util/admissioncheck/admissioncheck_test.go
@@ -18,21 +18,24 @@ package admissioncheck
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
-	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 )
 
 func TestConfigHelper(t *testing.T) {
-	ctx, _ := utiltesting.ContextWithLog(t)
 	testConfig := utiltesting.MakeProvisioningRequestConfig("config").
 		ProvisioningClass("className").
 		WithParameter("p1", "v1").
@@ -94,10 +97,7 @@ func TestConfigHelper(t *testing.T) {
 
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
-			scheme := runtime.NewScheme()
-			utilruntime.Must(clientgoscheme.AddToScheme(scheme))
-			utilruntime.Must(kueue.AddToScheme(scheme))
-			builder := fake.NewClientBuilder().WithScheme(scheme)
+			builder := utiltesting.NewClientBuilder()
 			if tc.admissioncheck != nil {
 				builder = builder.WithObjects(tc.admissioncheck)
 			}
@@ -105,8 +105,7 @@ func TestConfigHelper(t *testing.T) {
 				builder = builder.WithObjects(tc.config)
 			}
 			client := builder.Build()
-			ctx, cancel := context.WithCancel(ctx)
-			defer cancel()
+			ctx, _ := utiltesting.ContextWithLog(t)
 
 			helper, err := NewConfigHelper[*kueue.ProvisioningRequestConfig](client)
 
@@ -200,18 +199,12 @@ func TestFilterCheckStates(t *testing.T) {
 
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
-			scheme := runtime.NewScheme()
-			utilruntime.Must(clientgoscheme.AddToScheme(scheme))
-			utilruntime.Must(kueue.AddToScheme(scheme))
-			builder := fake.NewClientBuilder().WithScheme(scheme)
+			builder := utiltesting.NewClientBuilder()
 			if len(tc.admissionchecks) > 0 {
 				builder = builder.WithLists(&kueue.AdmissionCheckList{Items: tc.admissionchecks})
 			}
 			client := builder.Build()
 			ctx, _ := utiltesting.ContextWithLog(t)
-			ctx, cancel := context.WithCancel(ctx)
-			defer cancel()
-
 			gotResult, _ := FilterForController(ctx, client, tc.states, "test-controller")
 
 			if diff := cmp.Diff(tc.wantResult, gotResult); diff != "" {
@@ -220,3 +213,148 @@ func TestFilterCheckStates(t *testing.T) {
 		})
 	}
 }
+
+func TestGetMultiKueueAdmissionCheck(t *testing.T) {
+	cases := map[string]struct {
+		admissionChecks []kueue.AdmissionCheck
+		workloadACS     []kueue.AdmissionCheckState
+		wantResult      *kueue.AdmissionCheckState
+		wantErr         error
+	}{
+		"empty workload states": {
+			workloadACS: []kueue.AdmissionCheckState{},
+			wantResult:  nil,
+			wantErr:     nil,
+		},
+		"no relevant checks": {
+			admissionChecks: []kueue.AdmissionCheck{
+				*utiltesting.MakeAdmissionCheck("check1").ControllerName("other-controller").Obj(),
+				*utiltesting.MakeAdmissionCheck("check2").ControllerName("other-controller").Obj(),
+			},
+			workloadACS: []kueue.AdmissionCheckState{
+				{Name: "check1"},
+				{Name: "check2"},
+			},
+			wantResult: nil,
+			wantErr:    nil,
+		},
+		"one relevant check": {
+			admissionChecks: []kueue.AdmissionCheck{
+				*utiltesting.MakeAdmissionCheck("check1").ControllerName(kueue.MultiKueueControllerName).Obj(),
+				*utiltesting.MakeAdmissionCheck("check2").ControllerName("other-controller").Obj(),
+			},
+			workloadACS: []kueue.AdmissionCheckState{
+				{Name: "check1"},
+				{Name: "check2"},
+			},
+			wantResult: &kueue.AdmissionCheckState{Name: "check1"},
+			wantErr:    nil,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			builder := utiltesting.NewClientBuilder()
+			if len(tc.admissionChecks) > 0 {
+				builder = builder.WithLists(&kueue.AdmissionCheckList{Items: tc.admissionChecks})
+			}
+			client := builder.Build()
+			ctx, _ := utiltesting.ContextWithLog(t)
+
+			workload := &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					AdmissionChecks: tc.workloadACS,
+				},
+			}
+
+			gotResult, err := GetMultiKueueAdmissionCheck(ctx, client, workload)
+
+			if diff := cmp.Diff(tc.wantErr, err); diff != "" {
+				t.Errorf("unexpected error (-want/+got):\n%s", diff)
+			}
+
+			if diff := cmp.Diff(tc.wantResult, gotResult); diff != "" {
+				t.Errorf("unexpected result (-want/+got):\n%s", diff)
+			}
+		})
+	}
+}
+func TestGetRemoteClusters(t *testing.T) {
+	multiKueueConfigGetError := errors.New("failed to get MultiKueueConfig")
+	acTest := "test-admission-check"
+	cases := map[string]struct {
+		multiKueueConfig *kueue.MultiKueueConfig
+		wantResult       sets.Set[string]
+		wantErr          error
+	}{
+		"no clusters": {
+			multiKueueConfig: &kueue.MultiKueueConfig{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: acTest,
+				},
+				Spec: kueue.MultiKueueConfigSpec{
+					Clusters: []string{},
+				},
+			},
+			wantResult: nil,
+			wantErr:    ErrNoActiveClusters,
+		},
+		"multiple clusters": {
+			multiKueueConfig: &kueue.MultiKueueConfig{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: acTest,
+				},
+				Spec: kueue.MultiKueueConfigSpec{
+					Clusters: []string{"cluster1", "cluster2"},
+				},
+			},
+			wantResult: sets.New("cluster1", "cluster2"),
+		},
+		"error fetching config": {
+			multiKueueConfig: nil,
+			wantResult:       nil,
+			wantErr:          multiKueueConfigGetError,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			builder := utiltesting.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+				Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+					if _, isMKCfg := obj.(*kueue.MultiKueueConfig); isMKCfg && errors.Is(tc.wantErr, multiKueueConfigGetError) {
+						return tc.wantErr
+					}
+					return c.Get(ctx, key, obj, opts...)
+				},
+			})
+			ac := utiltesting.MakeAdmissionCheck(acTest).
+				ControllerName(kueue.MultiKueueControllerName).
+				Parameters(kueue.GroupVersion.Group, "MultiKueueConfig", acTest).
+				Obj()
+			builder = builder.WithObjects(ac)
+
+			if tc.multiKueueConfig != nil {
+				builder = builder.WithObjects(tc.multiKueueConfig)
+			}
+			client := builder.Build()
+
+			helper, err := NewMultiKueueStoreHelper(client)
+			if err != nil {
+				t.Fatalf("failed to create MultiKueueStoreHelper: %v", err)
+			}
+
+			ctx, _ := utiltesting.ContextWithLog(t)
+			gotResult, err := GetRemoteClusters(ctx, helper, kueue.AdmissionCheckReference(acTest))
+
+			if err != nil {
+				if diff := cmp.Diff(tc.wantErr, err, cmpopts.EquateErrors()); diff != "" {
+					t.Errorf("unexpected error (-want/+got):\n%s", diff)
+				}
+			}
+
+			if diff := cmp.Diff(tc.wantResult, gotResult); diff != "" {
+				t.Errorf("unexpected result (-want/+got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/pkg/util/testing/wrappers.go b/pkg/util/testing/wrappers.go
index 3e52eaf78..f51e7f47b 100644
--- a/pkg/util/testing/wrappers.go
+++ b/pkg/util/testing/wrappers.go
@@ -404,7 +404,7 @@ func (w *WorkloadWrapper) ClusterName(clusterName string) *WorkloadWrapper {
 	return w
 }
 
-func (w *WorkloadWrapper) NominatedClusterNames(nominatedClusterNames []string) *WorkloadWrapper {
+func (w *WorkloadWrapper) NominatedClusterNames(nominatedClusterNames ...string) *WorkloadWrapper {
 	w.Status.NominatedClusterNames = nominatedClusterNames
 	return w
 }
diff --git a/test/integration/multikueue/suite_test.go b/test/integration/multikueue/suite_test.go
index 354004946..641ea6dd4 100644
--- a/test/integration/multikueue/suite_test.go
+++ b/test/integration/multikueue/suite_test.go
@@ -53,6 +53,7 @@ import (
 	workloadpod "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
 	workloadraycluster "sigs.k8s.io/kueue/pkg/controller/jobs/raycluster"
 	workloadrayjob "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob"
+	dispatcher "sigs.k8s.io/kueue/pkg/controller/workloaddispatcher"
 	"sigs.k8s.io/kueue/pkg/util/kubeversion"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/webhooks"
@@ -297,6 +298,11 @@ func managerAndMultiKueueSetup(
 		multikueue.WithDispatcherName(dispatcherName),
 	)
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+	configuration := &config.Configuration{}
+	mgr.GetScheme().Default(configuration)
+	_, err = dispatcher.SetupControllers(mgr, configuration, dispatcherName)
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 }
 
 var _ = ginkgo.BeforeSuite(func() {