Repository: kubernetes-sigs/kueue

Issue:
Replace sort.Slice() with slices.Sort()
<!-- Please only use this template for submitting clean up requests -->

**What would you like to be cleaned**:

I would propose to replace `sort.Slice()` with `slices.Sort()` everywhere.

**Why is this needed**:

Both are implemented with the QuickSort base algorithm. But the new `slices.Sort()` function is faster than `sort.Slice()` in many situations as described in Go doc:

> Note: in many situations, the newer [slices.IsSortedFunc](https://pkg.go.dev/slices#IsSortedFunc) function is more ergonomic and runs faster.

https://pkg.go.dev/sort#Slice

Hints from issue comments:
@tenzen-y: /assign @wzshiming

Solution patch:
diff --git a/.golangci.yaml b/.golangci.yaml
index b090b5d66..5c873a09c 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -21,6 +21,7 @@ linters:
     - perfsprint
     - revive
     - unconvert
+    - forbidigo
   settings:
     gocritic:
       enabled-checks:
@@ -70,6 +71,10 @@ linters:
         - name: var-naming
         - name: redundant-import-alias
         - name: use-any
+    forbidigo:
+      forbid:
+        - pattern: ^sort\.Slice$
+          msg: Please use slices.SortFunc instead of sort.Slice for better performance and consistency
   exclusions:
     generated: lax
     presets:
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/cache/queue/cluster_queue.go b/pkg/cache/queue/cluster_queue.go
index ae9ba40f1..a38c1b305 100644
--- a/pkg/cache/queue/cluster_queue.go
+++ b/pkg/cache/queue/cluster_queue.go
@@ -18,7 +18,7 @@ package queue
 
 import (
 	"context"
-	"sort"
+	"slices"
 	"sync"
 
 	corev1 "k8s.io/api/core/v1"
@@ -404,8 +404,11 @@ func (c *ClusterQueue) DumpInadmissible() ([]workload.Reference, bool) {
 // this ClusterQueue.
 func (c *ClusterQueue) Snapshot() []*workload.Info {
 	elements := c.totalElements()
-	sort.Slice(elements, func(i, j int) bool {
-		return c.lessFunc(elements[i], elements[j])
+	slices.SortFunc(elements, func(a, b *workload.Info) int {
+		if c.lessFunc(a, b) {
+			return -1
+		}
+		return 1
 	})
 	return elements
 }
diff --git a/pkg/cache/scheduler/cache.go b/pkg/cache/scheduler/cache.go
index eddc25245..8234b0894 100644
--- a/pkg/cache/scheduler/cache.go
+++ b/pkg/cache/scheduler/cache.go
@@ -17,10 +17,11 @@ limitations under the License.
 package scheduler
 
 import (
+	"cmp"
 	"context"
 	"errors"
 	"fmt"
-	"sort"
+	"slices"
 	"sync"
 
 	"github.com/go-logr/logr"
@@ -790,8 +791,8 @@ func getUsage(frq resources.FlavorResourceQuantities, cq *clusterQueue) []kueue.
 				outFlvUsage.Resources = append(outFlvUsage.Resources, rUsage)
 			}
 			// The resourceUsages should be in a stable order to avoid endless creation of update events.
-			sort.Slice(outFlvUsage.Resources, func(i, j int) bool {
-				return outFlvUsage.Resources[i].Name < outFlvUsage.Resources[j].Name
+			slices.SortFunc(outFlvUsage.Resources, func(a, b kueue.ResourceUsage) int {
+				return cmp.Compare(a.Name, b.Name)
 			})
 			usage = append(usage, outFlvUsage)
 		}
@@ -885,8 +886,8 @@ func filterLocalQueueUsage(orig resources.FlavorResourceQuantities, resourceGrou
 				})
 			}
 			// The resourceUsages should be in a stable order to avoid endless creation of update events.
-			sort.Slice(outFlvUsage.Resources, func(i, j int) bool {
-				return outFlvUsage.Resources[i].Name < outFlvUsage.Resources[j].Name
+			slices.SortFunc(outFlvUsage.Resources, func(a, b kueue.LocalQueueResourceUsage) int {
+				return cmp.Compare(a.Name, b.Name)
 			})
 			qFlvUsages = append(qFlvUsages, outFlvUsage)
 		}
diff --git a/pkg/cache/scheduler/tas_flavor_snapshot.go b/pkg/cache/scheduler/tas_flavor_snapshot.go
index eefc4a8ab..02d69fe04 100644
--- a/pkg/cache/scheduler/tas_flavor_snapshot.go
+++ b/pkg/cache/scheduler/tas_flavor_snapshot.go
@@ -24,7 +24,6 @@ import (
 	"maps"
 	"math"
 	"slices"
-	"sort"
 	"strconv"
 	"strings"
 
@@ -775,10 +774,10 @@ func (s *TASFlavorSnapshot) mergeTopologyAssignments(a, b *kueue.TopologyAssignm
 	sortedDomains := make([]kueue.TopologyDomainAssignment, 0, len(a.Domains)+len(b.Domains))
 	sortedDomains = append(sortedDomains, a.Domains...)
 	sortedDomains = append(sortedDomains, b.Domains...)
-	sort.Slice(sortedDomains, func(i, j int) bool {
-		a, b := sortedDomains[i], sortedDomains[j]
-		aDomain, bDomain := s.domainsPerLevel[nodeLevel][utiltas.DomainID(a.Values)], s.domainsPerLevel[nodeLevel][utiltas.DomainID(b.Values)]
-		return utiltas.DomainID(aDomain.levelValues) < utiltas.DomainID(bDomain.levelValues)
+	slices.SortFunc(sortedDomains, func(a, b kueue.TopologyDomainAssignment) int {
+		aDomain := s.domainsPerLevel[nodeLevel][utiltas.DomainID(a.Values)]
+		bDomain := s.domainsPerLevel[nodeLevel][utiltas.DomainID(b.Values)]
+		return cmp.Compare(utiltas.DomainID(aDomain.levelValues), utiltas.DomainID(bDomain.levelValues))
 	})
 	mergedDomains := make([]kueue.TopologyDomainAssignment, 0, len(sortedDomains))
 	for _, domain := range sortedDomains {
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/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index e61e09ffa..a4675142f 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -364,7 +364,7 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 				log.Error(err, "couldn't get an ancestor job workload")
 				return ctrl.Result{}, err
 			} else if ancestorWorkload == nil || !workload.IsAdmitted(ancestorWorkload) {
-				if err := clientutil.Patch(ctx, r.client, object, true, func() (client.Object, bool, error) {
+				if err := clientutil.Patch(ctx, r.client, object, func() (client.Object, bool, error) {
 					job.Suspend()
 					return object, true, nil
 				}); err != nil {
@@ -1085,7 +1085,7 @@ func (r *JobReconciler) startJob(ctx context.Context, job GenericJob, object cli
 			return err
 		}
 	} else {
-		if err := clientutil.Patch(ctx, r.client, object, true, func() (client.Object, bool, error) {
+		if err := clientutil.Patch(ctx, r.client, object, func() (client.Object, bool, error) {
 			return object, true, job.RunWithPodSetsInfo(info)
 		}); err != nil {
 			return err
@@ -1129,7 +1129,7 @@ func (r *JobReconciler) stopJob(ctx context.Context, job GenericJob, wl *kueue.W
 		return nil
 	}
 
-	if err := clientutil.Patch(ctx, r.client, object, true, func() (client.Object, bool, error) {
+	if err := clientutil.Patch(ctx, r.client, object, func() (client.Object, bool, error) {
 		job.Suspend()
 		if info != nil {
 			job.RestorePodSetsInfo(info)
diff --git a/pkg/controller/jobs/job/job_controller.go b/pkg/controller/jobs/job/job_controller.go
index dc47d149a..1e8ca4b21 100644
--- a/pkg/controller/jobs/job/job_controller.go
+++ b/pkg/controller/jobs/job/job_controller.go
@@ -167,7 +167,7 @@ func (j *Job) Stop(ctx context.Context, c client.Client, podSetsInfo []podset.Po
 	stoppedNow := false
 
 	if !j.IsSuspended() {
-		if err := clientutil.Patch(ctx, c, object, true, func() (client.Object, bool, error) {
+		if err := clientutil.Patch(ctx, c, object, func() (client.Object, bool, error) {
 			j.Suspend()
 			if j.Annotations == nil {
 				j.Annotations = map[string]string{}
@@ -191,7 +191,7 @@ func (j *Job) Stop(ctx context.Context, c client.Client, podSetsInfo []podset.Po
 		}
 	}
 
-	if err := clientutil.Patch(ctx, c, object, true, func() (client.Object, bool, error) {
+	if err := clientutil.Patch(ctx, c, object, func() (client.Object, bool, error) {
 		j.RestorePodSetsInfo(podSetsInfo)
 		delete(j.Annotations, StoppingAnnotation)
 		return object, true, nil
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
index b5036ceb9..27a93326d 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
@@ -72,7 +72,7 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 	log.V(2).Info("Reconcile LeaderWorkerSet Pod")
 
 	if utilpod.IsTerminated(pod) {
-		err = client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, true, func() (client.Object, bool, error) {
+		err = client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, func() (client.Object, bool, error) {
 			removed := controllerutil.RemoveFinalizer(pod, podconstants.PodFinalizer)
 			if removed {
 				log.V(3).Info(
@@ -85,7 +85,7 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 			return pod, removed, nil
 		}))
 	} else {
-		err = client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, true, func() (client.Object, bool, error) {
+		err = client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, func() (client.Object, bool, error) {
 			updated, err := r.setDefault(ctx, pod)
 			if err != nil {
 				return nil, false, err
diff --git a/pkg/controller/jobs/mpijob/mpijob_webhook.go b/pkg/controller/jobs/mpijob/mpijob_webhook.go
index 56f76d591..c82edec06 100644
--- a/pkg/controller/jobs/mpijob/mpijob_webhook.go
+++ b/pkg/controller/jobs/mpijob/mpijob_webhook.go
@@ -17,8 +17,9 @@ limitations under the License.
 package mpijob
 
 import (
+	"cmp"
 	"context"
-	"sort"
+	"slices"
 
 	"github.com/kubeflow/mpi-operator/pkg/apis/kubeflow/v2beta1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -103,8 +104,8 @@ func (w *MpiJobWebhook) ValidateCreate(ctx context.Context, obj runtime.Object)
 	if err != nil {
 		return nil, err
 	}
-	sort.Slice(validationErrs, func(i, j int) bool {
-		return validationErrs[i].Field < validationErrs[j].Field
+	slices.SortFunc(validationErrs, func(a, b *field.Error) int {
+		return cmp.Compare(a.Field, b.Field)
 	})
 	return nil, validationErrs.ToAggregate()
 }
@@ -121,8 +122,8 @@ func (w *MpiJobWebhook) ValidateUpdate(ctx context.Context, oldObj, newObj runti
 		return nil, err
 	}
 	allErrs = append(allErrs, validationErrs...)
-	sort.Slice(validationErrs, func(i, j int) bool {
-		return validationErrs[i].Field < validationErrs[j].Field
+	slices.SortFunc(validationErrs, func(a, b *field.Error) int {
+		return cmp.Compare(a.Field, b.Field)
 	})
 	return nil, allErrs.ToAggregate()
 }
diff --git a/pkg/controller/jobs/pod/pod_controller.go b/pkg/controller/jobs/pod/pod_controller.go
index d3b143df1..7c6a88a1e 100644
--- a/pkg/controller/jobs/pod/pod_controller.go
+++ b/pkg/controller/jobs/pod/pod_controller.go
@@ -22,7 +22,6 @@ import (
 	"errors"
 	"fmt"
 	"slices"
-	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -52,6 +51,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/podset"
 	clientutil "sigs.k8s.io/kueue/pkg/util/client"
+	cmputil "sigs.k8s.io/kueue/pkg/util/cmp"
 	"sigs.k8s.io/kueue/pkg/util/expectations"
 	utilmaps "sigs.k8s.io/kueue/pkg/util/maps"
 	"sigs.k8s.io/kueue/pkg/util/parallelize"
@@ -265,7 +265,7 @@ func (p *Pod) Run(ctx context.Context, c client.Client, podSetsInfo []podset.Pod
 			return nil
 		}
 
-		if err := clientutil.Patch(ctx, c, &p.pod, true, func() (client.Object, bool, error) {
+		if err := clientutil.Patch(ctx, c, &p.pod, func() (client.Object, bool, error) {
 			return &p.pod, true, prepare(&p.pod, podSetsInfo[0])
 		}); err != nil {
 			return err
@@ -285,7 +285,7 @@ func (p *Pod) Run(ctx context.Context, c client.Client, podSetsInfo []podset.Pod
 			return nil
 		}
 
-		if err := clientutil.Patch(ctx, c, pod, true, func() (client.Object, bool, error) {
+		if err := clientutil.Patch(ctx, c, pod, func() (client.Object, bool, error) {
 			roleHash, err := getRoleHash(*pod)
 			if err != nil {
 				return nil, false, err
@@ -529,9 +529,9 @@ func (p *Pod) Finalize(ctx context.Context, c client.Client) error {
 
 	return parallelize.Until(ctx, len(podsInGroup.Items), func(i int) error {
 		pod := &podsInGroup.Items[i]
-		return clientutil.Patch(ctx, c, pod, false, func() (client.Object, bool, error) {
+		return clientutil.Patch(ctx, c, pod, func() (client.Object, bool, error) {
 			return pod, controllerutil.RemoveFinalizer(pod, podconstants.PodFinalizer), nil
-		})
+		}, clientutil.WithNonStrict())
 	})
 }
 
@@ -827,22 +827,21 @@ func lastActiveTime(clock clock.Clock, p *corev1.Pod) time.Time {
 // - lastActiveTime (pods that were active last are first)
 // - creation timestamp (newer pods are first)
 func sortInactivePods(clock clock.Clock, inactivePods []corev1.Pod) {
-	sort.Slice(inactivePods, func(i, j int) bool {
-		pi := &inactivePods[i]
-		pj := &inactivePods[j]
-		iFin := slices.Contains(pi.Finalizers, podconstants.PodFinalizer)
-		jFin := slices.Contains(pj.Finalizers, podconstants.PodFinalizer)
-		if iFin != jFin {
-			return iFin
-		}
-
-		iLastActive := lastActiveTime(clock, pi)
-		jLastActive := lastActiveTime(clock, pj)
-
-		if iLastActive.Equal(jLastActive) {
-			return pi.CreationTimestamp.Before(&pj.CreationTimestamp)
-		}
-		return jLastActive.Before(iLastActive)
+	slices.SortFunc(inactivePods, func(pi, pj corev1.Pod) int {
+		return cmputil.LazyOr(
+			func() int {
+				return cmputil.CompareBool(
+					slices.Contains(pi.Finalizers, podconstants.PodFinalizer),
+					slices.Contains(pj.Finalizers, podconstants.PodFinalizer),
+				)
+			},
+			func() int {
+				return lastActiveTime(clock, &pj).Compare(lastActiveTime(clock, &pi))
+			},
+			func() int {
+				return pi.CreationTimestamp.Compare(pj.CreationTimestamp.Time)
+			},
+		)
 	})
 }
 
@@ -852,22 +851,26 @@ func sortInactivePods(clock clock.Clock, inactivePods []corev1.Pod) {
 // - creation timestamp (newer pods are last)
 func sortActivePods(activePods []corev1.Pod) {
 	// Sort active pods by creation timestamp
-	sort.Slice(activePods, func(i, j int) bool {
-		pi := &activePods[i]
-		pj := &activePods[j]
-		iFin := slices.Contains(pi.Finalizers, podconstants.PodFinalizer)
-		jFin := slices.Contains(pj.Finalizers, podconstants.PodFinalizer)
-		// Prefer to keep pods that have a finalizer.
-		if iFin != jFin {
-			return iFin
-		}
-		iGated := isGated(pi)
-		jGated := isGated(pj)
-		// Prefer to keep pods that aren't gated.
-		if iGated != jGated {
-			return !iGated
-		}
-		return pi.CreationTimestamp.Before(&pj.CreationTimestamp)
+	slices.SortFunc(activePods, func(pi, pj corev1.Pod) int {
+		return cmputil.LazyOr(
+			func() int {
+				// Prefer to keep pods that have a finalizer.
+				return cmputil.CompareBool(
+					slices.Contains(pi.Finalizers, podconstants.PodFinalizer),
+					slices.Contains(pj.Finalizers, podconstants.PodFinalizer),
+				)
+			},
+			func() int {
+				// Prefer to keep pods that aren't gated.
+				return cmputil.CompareBool(
+					isGated(&pj),
+					isGated(&pi),
+				)
+			},
+			func() int {
+				return pi.CreationTimestamp.Compare(pj.CreationTimestamp.Time)
+			},
+		)
 	})
 }
 
@@ -885,13 +888,13 @@ func (p *Pod) removeExcessPods(ctx context.Context, c client.Client, r record.Ev
 	// Finalize and delete the active pods created last
 	err := parallelize.Until(ctx, len(extraPods), func(i int) error {
 		pod := extraPods[i]
-		if err := clientutil.Patch(ctx, c, &pod, false, func() (client.Object, bool, error) {
+		if err := clientutil.Patch(ctx, c, &pod, func() (client.Object, bool, error) {
 			removed := controllerutil.RemoveFinalizer(&pod, podconstants.PodFinalizer)
 			if removed {
 				log.V(3).Info("Finalizing excess pod in group", "excessPod", klog.KObj(&pod))
 			}
 			return &pod, removed, nil
-		}); err != nil {
+		}, clientutil.WithNonStrict()); err != nil {
 			// We won't observe this cleanup in the event handler.
 			p.excessPodExpectations.ObservedUID(log, p.key, pod.UID)
 			return err
@@ -927,13 +930,13 @@ func (p *Pod) finalizePods(ctx context.Context, c client.Client, extraPods []cor
 	err := parallelize.Until(ctx, len(extraPods), func(i int) error {
 		pod := extraPods[i]
 		var removed bool
-		if err := clientutil.Patch(ctx, c, &pod, false, func() (client.Object, bool, error) {
+		if err := clientutil.Patch(ctx, c, &pod, func() (client.Object, bool, error) {
 			removed = controllerutil.RemoveFinalizer(&pod, podconstants.PodFinalizer)
 			if removed {
 				log.V(3).Info("Finalizing pod in group", "Pod", klog.KObj(&pod))
 			}
 			return &pod, removed, nil
-		}); err != nil {
+		}, clientutil.WithNonStrict()); err != nil {
 			// We won't observe this cleanup in the event handler.
 			p.excessPodExpectations.ObservedUID(log, p.key, pod.UID)
 			return err
diff --git a/pkg/controller/jobs/statefulset/statefulset_reconciler.go b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
index ba53b49f7..9847b5ad1 100644
--- a/pkg/controller/jobs/statefulset/statefulset_reconciler.go
+++ b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
@@ -103,7 +103,7 @@ func (r *Reconciler) finalizePods(ctx context.Context, sts *appsv1.StatefulSet,
 
 func (r *Reconciler) finalizePod(ctx context.Context, sts *appsv1.StatefulSet, pod *corev1.Pod) error {
 	log := ctrl.LoggerFrom(ctx)
-	return client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, true, func() (client.Object, bool, error) {
+	return client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, func() (client.Object, bool, error) {
 		if ungateAndFinalize(sts, pod) {
 			log.V(3).Info(
 				"Finalizing pod in group",
diff --git a/pkg/controller/tas/topology_ungater.go b/pkg/controller/tas/topology_ungater.go
index 6bc2b53cb..242ef947a 100644
--- a/pkg/controller/tas/topology_ungater.go
+++ b/pkg/controller/tas/topology_ungater.go
@@ -240,7 +240,7 @@ func (r *topologyUngater) Reconcile(ctx context.Context, req reconcile.Request)
 	err := parallelize.Until(ctx, len(allToUngate), func(i int) error {
 		podWithUngateInfo := &allToUngate[i]
 		var ungated bool
-		e := utilclient.Patch(ctx, r.client, podWithUngateInfo.pod, true, func() (client.Object, bool, error) {
+		e := utilclient.Patch(ctx, r.client, podWithUngateInfo.pod, func() (client.Object, bool, error) {
 			log.V(3).Info("ungating pod", "pod", klog.KObj(podWithUngateInfo.pod), "nodeLabels", podWithUngateInfo.nodeLabels)
 			ungated = utilpod.Ungate(podWithUngateInfo.pod, kueuealpha.TopologySchedulingGate)
 			if podWithUngateInfo.pod.Spec.NodeSelector == nil {
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/scheduler/preemption/classical/candidate_generator.go b/pkg/scheduler/preemption/classical/candidate_generator.go
index 240683edb..3e583c32d 100644
--- a/pkg/scheduler/preemption/classical/candidate_generator.go
+++ b/pkg/scheduler/preemption/classical/candidate_generator.go
@@ -17,6 +17,7 @@ limitations under the License.
 package classical
 
 import (
+	"slices"
 	"sort"
 	"time"
 
@@ -80,18 +81,18 @@ func NewCandidateIterator(
 	frsNeedPreemption sets.Set[resources.FlavorResource],
 	snapshot *schdcache.Snapshot,
 	clock clock.Clock,
-	ordering func(logr.Logger, bool, *workload.Info, *workload.Info, kueue.ClusterQueueReference, time.Time) bool,
+	ordering func(logr.Logger, bool, *workload.Info, *workload.Info, kueue.ClusterQueueReference, time.Time) int,
 ) *candidateIterator {
 	sameQueueCandidates := collectSameQueueCandidates(hierarchicalReclaimCtx)
 	hierarchyCandidates, priorityCandidates := collectCandidatesForHierarchicalReclaim(hierarchicalReclaimCtx)
-	sort.Slice(sameQueueCandidates, func(i, j int) bool {
-		return ordering(hierarchicalReclaimCtx.Log, enabledAfs, sameQueueCandidates[i].wl, sameQueueCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
+	slices.SortFunc(sameQueueCandidates, func(a, b *candidateElem) int {
+		return ordering(hierarchicalReclaimCtx.Log, enabledAfs, a.wl, b.wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
 	})
-	sort.Slice(priorityCandidates, func(i, j int) bool {
-		return ordering(hierarchicalReclaimCtx.Log, enabledAfs, priorityCandidates[i].wl, priorityCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
+	slices.SortFunc(priorityCandidates, func(a, b *candidateElem) int {
+		return ordering(hierarchicalReclaimCtx.Log, enabledAfs, a.wl, b.wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
 	})
-	sort.Slice(hierarchyCandidates, func(i, j int) bool {
-		return ordering(hierarchicalReclaimCtx.Log, enabledAfs, hierarchyCandidates[i].wl, hierarchyCandidates[j].wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
+	slices.SortFunc(hierarchyCandidates, func(a, b *candidateElem) int {
+		return ordering(hierarchicalReclaimCtx.Log, enabledAfs, a.wl, b.wl, hierarchicalReclaimCtx.Cq.Name, clock.Now())
 	})
 
 	evictedHierarchicalReclaimCandidates, nonEvictedHierarchicalReclaimCandidates := splitEvicted(hierarchyCandidates)
diff --git a/pkg/scheduler/preemption/common/ordering.go b/pkg/scheduler/preemption/common/ordering.go
index 3c34ea3d6..c316f738d 100644
--- a/pkg/scheduler/preemption/common/ordering.go
+++ b/pkg/scheduler/preemption/common/ordering.go
@@ -17,6 +17,7 @@ limitations under the License.
 package preemptioncommon
 
 import (
+	"cmp"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -25,6 +26,7 @@ import (
 	"k8s.io/klog/v2"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	cmputil "sigs.k8s.io/kueue/pkg/util/cmp"
 	"sigs.k8s.io/kueue/pkg/util/priority"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
@@ -36,38 +38,48 @@ import (
 // 2. (AdmissionFairSharing only) Workloads with lower LocalQueue's usage first
 // 3. Workloads with lower priority first.
 // 4. Workloads admitted more recently first.
-func CandidatesOrdering(log logr.Logger, afsEnabled bool, a, b *workload.Info, cq kueue.ClusterQueueReference, now time.Time) bool {
-	aEvicted := meta.IsStatusConditionTrue(a.Obj.Status.Conditions, kueue.WorkloadEvicted)
-	bEvicted := meta.IsStatusConditionTrue(b.Obj.Status.Conditions, kueue.WorkloadEvicted)
-	if aEvicted != bEvicted {
-		return aEvicted
-	}
-	aInCQ := a.ClusterQueue == cq
-	bInCQ := b.ClusterQueue == cq
-	if aInCQ != bInCQ {
-		return !aInCQ
-	}
-
-	if afsEnabled && resourceUsagePreemptionEnabled(a, b) {
-		if a.LocalQueueFSUsage != b.LocalQueueFSUsage {
-			log.V(3).Info("Comparing workloads by LocalQueue fair sharing usage",
-				"workloadA", klog.KObj(a.Obj), "queueA", a.Obj.Spec.QueueName, "usageA", a.LocalQueueFSUsage,
-				"workloadB", klog.KObj(b.Obj), "queueB", b.Obj.Spec.QueueName, "usageB", b.LocalQueueFSUsage)
-			return *a.LocalQueueFSUsage > *b.LocalQueueFSUsage
-		}
-	}
-	pa := priority.Priority(a.Obj)
-	pb := priority.Priority(b.Obj)
-	if pa != pb {
-		return pa < pb
-	}
-	timeA := quotaReservationTime(a.Obj, now)
-	timeB := quotaReservationTime(b.Obj, now)
-	if !timeA.Equal(timeB) {
-		return timeA.After(timeB)
-	}
-	// Arbitrary comparison for deterministic sorting.
-	return a.Obj.UID < b.Obj.UID
+func CandidatesOrdering(log logr.Logger, afsEnabled bool, a, b *workload.Info, cq kueue.ClusterQueueReference, now time.Time) int {
+	return cmputil.LazyOr(
+		func() int {
+			return cmputil.CompareBool(
+				meta.IsStatusConditionTrue(a.Obj.Status.Conditions, kueue.WorkloadEvicted),
+				meta.IsStatusConditionTrue(b.Obj.Status.Conditions, kueue.WorkloadEvicted),
+			)
+		},
+		func() int {
+			return cmputil.CompareBool(
+				b.ClusterQueue == cq,
+				a.ClusterQueue == cq,
+			)
+		},
+		func() int {
+			if afsEnabled &&
+				resourceUsagePreemptionEnabled(a, b) &&
+				a.LocalQueueFSUsage != b.LocalQueueFSUsage {
+				log.V(3).Info("Comparing workloads by LocalQueue fair sharing usage",
+					"workloadA", klog.KObj(a.Obj), "queueA", a.Obj.Spec.QueueName, "usageA", a.LocalQueueFSUsage,
+					"workloadB", klog.KObj(b.Obj), "queueB", b.Obj.Spec.QueueName, "usageB", b.LocalQueueFSUsage)
+				return cmp.Compare(*b.LocalQueueFSUsage, *a.LocalQueueFSUsage)
+			}
+			return 0
+		},
+		func() int {
+			return cmp.Compare(
+				priority.Priority(a.Obj),
+				priority.Priority(b.Obj),
+			)
+		},
+		func() int {
+			return quotaReservationTime(b.Obj, now).Compare(quotaReservationTime(a.Obj, now))
+		},
+		func() int {
+			// Arbitrary comparison for deterministic sorting.
+			return cmp.Compare(
+				a.Obj.UID,
+				b.Obj.UID,
+			)
+		},
+	)
 }
 
 func resourceUsagePreemptionEnabled(a, b *workload.Info) bool {
diff --git a/pkg/scheduler/preemption/fairsharing/ordering.go b/pkg/scheduler/preemption/fairsharing/ordering.go
index da6558e47..decc9dafa 100644
--- a/pkg/scheduler/preemption/fairsharing/ordering.go
+++ b/pkg/scheduler/preemption/fairsharing/ordering.go
@@ -152,7 +152,7 @@ func (t *TargetClusterQueueOrdering) nextTarget(cohort *schdcache.CohortSnapshot
 		case drs == highestCqDrs:
 			newCandWl := t.clusterQueueToTarget[cq.GetName()][0]
 			currentCandWl := t.clusterQueueToTarget[highestCq.GetName()][0]
-			if preemptioncommon.CandidatesOrdering(t.log, false, newCandWl, currentCandWl, t.preemptorCq.Name, time.Now()) {
+			if preemptioncommon.CandidatesOrdering(t.log, false, newCandWl, currentCandWl, t.preemptorCq.Name, time.Now()) < 0 {
 				highestCq = cq
 			}
 		case drs > highestCqDrs:
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 89d102f66..6030bdd7a 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -19,7 +19,7 @@ package preemption
 import (
 	"context"
 	"fmt"
-	"sort"
+	"slices"
 	"sync/atomic"
 
 	"github.com/go-logr/logr"
@@ -386,8 +386,8 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, strategies []f
 	if len(candidates) == 0 {
 		return nil
 	}
-	sort.Slice(candidates, func(i, j int) bool {
-		return preemptioncommon.CandidatesOrdering(preemptionCtx.log, p.enabledAfs, candidates[i], candidates[j], preemptionCtx.preemptorCQ.Name, p.clock.Now())
+	slices.SortFunc(candidates, func(a, b *workload.Info) int {
+		return preemptioncommon.CandidatesOrdering(preemptionCtx.log, p.enabledAfs, a, b, preemptionCtx.preemptorCQ.Name, p.clock.Now())
 	})
 	if logV := preemptionCtx.log.V(5); logV.Enabled() {
 		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
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
+}
diff --git a/pkg/util/client/client.go b/pkg/util/client/client.go
index e614242b1..d09e23315 100644
--- a/pkg/util/client/client.go
+++ b/pkg/util/client/client.go
@@ -22,31 +22,49 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func createPatch(before, after client.Object) (client.Patch, error) {
-	patchBase := client.MergeFrom(before)
-	patchBytes, err := patchBase.Data(after)
-	if err != nil {
-		return nil, err
+type ClientPatchOption func(*ClientPatchOptions)
+
+type ClientPatchOptions struct {
+	NonStrict bool
+}
+
+func DefaultClientPatchOptions() *ClientPatchOptions {
+	return &ClientPatchOptions{
+		NonStrict: false, // default is strict
+	}
+}
+
+func WithNonStrict() ClientPatchOption {
+	return func(o *ClientPatchOptions) {
+		o.NonStrict = true
 	}
-	return client.RawPatch(patchBase.Type(), patchBytes), nil
+}
+
+func patchCommon(obj client.Object, update func() (client.Object, bool, error), patchFunc func(patch client.Patch) error, options ...ClientPatchOption) error {
+	opts := DefaultClientPatchOptions()
+	for _, opt := range options {
+		opt(opts)
+	}
+	strict := !opts.NonStrict
+	return updateAndPatch(obj, strict, update, patchFunc)
 }
 
 // Patch applies the merge patch of client.Object.
 // If strict is true, the resourceVersion will be part of the patch, make this call fail if
 // client.Object was changed.
-func Patch(ctx context.Context, c client.Client, obj client.Object, strict bool, update func() (client.Object, bool, error)) error {
-	return updateAndPatch(obj, strict, update, func(patch client.Patch) error {
+func Patch(ctx context.Context, c client.Client, obj client.Object, update func() (client.Object, bool, error), options ...ClientPatchOption) error {
+	return patchCommon(obj, update, func(patch client.Patch) error {
 		return c.Patch(ctx, obj, patch)
-	})
+	}, options...)
 }
 
 // PatchStatus applies the merge patch of client.Object status.
 // The resourceVersion will be part of the patch, make this call fail if
 // client.Object was changed.
-func PatchStatus(ctx context.Context, c client.Client, obj client.Object, update func() (client.Object, bool, error)) error {
-	return updateAndPatch(obj, true, update, func(patch client.Patch) error {
+func PatchStatus(ctx context.Context, c client.Client, obj client.Object, update func() (client.Object, bool, error), options ...ClientPatchOption) error {
+	return patchCommon(obj, update, func(patch client.Patch) error {
 		return c.Status().Patch(ctx, obj, patch)
-	})
+	}, options...)
 }
 
 func getOriginalObject(obj client.Object, strict bool) client.Object {
@@ -70,3 +88,12 @@ func updateAndPatch(obj client.Object, strict bool, update func() (client.Object
 	}
 	return patchFn(patch)
 }
+
+func createPatch(before, after client.Object) (client.Patch, error) {
+	patchBase := client.MergeFrom(before)
+	patchBytes, err := patchBase.Data(after)
+	if err != nil {
+		return nil, err
+	}
+	return client.RawPatch(patchBase.Type(), patchBytes), nil
+}
diff --git a/pkg/util/cmp/cmp.go b/pkg/util/cmp/cmp.go
new file mode 100644
index 000000000..7ff5ceee3
--- /dev/null
+++ b/pkg/util/cmp/cmp.go
@@ -0,0 +1,47 @@
+/*
+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 cmp
+
+// CompareBool compares two boolean values and returns:
+// - 0 if they are equal
+// - -1 if a is true and b is false
+// - 1 if a is false and b is true
+func CompareBool(a, b bool) int {
+	if a == b {
+		return 0
+	}
+
+	if a {
+		return -1
+	}
+
+	return 1
+}
+
+// LazyOr evaluates the provided functions in order and returns the first non-zero value.
+// It returns the zero value if all functions return zero values.
+// This is useful for implementing short-circuit evaluation in comparison functions.
+func LazyOr[T comparable](funcs ...func() T) T {
+	var zero T
+	for _, fn := range funcs {
+		n := fn()
+		if n != zero {
+			return n
+		}
+	}
+	return zero
+}
diff --git a/pkg/workloadslicing/workloadslicing.go b/pkg/workloadslicing/workloadslicing.go
index c02ac470f..3b21088a1 100644
--- a/pkg/workloadslicing/workloadslicing.go
+++ b/pkg/workloadslicing/workloadslicing.go
@@ -22,7 +22,6 @@ import (
 	"errors"
 	"fmt"
 	"slices"
-	"sort"
 
 	corev1 "k8s.io/api/core/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
@@ -36,6 +35,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/scheduler/preemption"
 	clientutil "sigs.k8s.io/kueue/pkg/util/client"
+	cmputil "sigs.k8s.io/kueue/pkg/util/cmp"
 	"sigs.k8s.io/kueue/pkg/util/pod"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
@@ -125,12 +125,18 @@ func FindNotFinishedWorkloads(ctx context.Context, clnt client.Client, jobObject
 	// as a tiebreaker. This edge case is uncommon in production but can occur in
 	// integration or e2e tests where the original and scaled-up workloads are created
 	// in rapid succession.
-	sort.Slice(list.Items, func(i, j int) bool {
-		a, b := list.Items[i], list.Items[j]
-		if a.CreationTimestamp.Equal(&b.CreationTimestamp) {
-			return b.Annotations[WorkloadSliceReplacementFor] == string(workload.Key(&a))
-		}
-		return a.CreationTimestamp.Before(&b.CreationTimestamp)
+	slices.SortFunc(list.Items, func(a, b kueue.Workload) int {
+		return cmputil.LazyOr(
+			func() int {
+				return a.CreationTimestamp.Compare(b.CreationTimestamp.Time)
+			},
+			func() int {
+				if b.Annotations[WorkloadSliceReplacementFor] == string(workload.Key(&a)) {
+					return -1
+				}
+				return 1
+			},
+		)
 	})
 
 	// Filter out workloads with activated "Finished" condition.
@@ -263,7 +269,7 @@ func StartWorkloadSlicePods(ctx context.Context, clnt client.Client, object clie
 		return fmt.Errorf("failed to list job pods: %w", err)
 	}
 	for i := range list.Items {
-		if err := clientutil.Patch(ctx, clnt, &list.Items[i], true, func() (client.Object, bool, error) {
+		if err := clientutil.Patch(ctx, clnt, &list.Items[i], func() (client.Object, bool, error) {
 			return &list.Items[i], pod.Ungate(&list.Items[i], kueue.ElasticJobSchedulingGate), nil
 		}); err != nil {
 			return fmt.Errorf("failed to patch pod: %w", err)



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/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index 13c3b8368..28074c01a 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -19,7 +19,7 @@ package preemption
 import (
 	"context"
 	"fmt"
-	"sort"
+	"slices"
 	"sync"
 	"testing"
 	"time"
@@ -46,7 +46,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/resources"
 	"sigs.k8s.io/kueue/pkg/scheduler/flavorassigner"
 	preemptioncommon "sigs.k8s.io/kueue/pkg/scheduler/preemption/common"
-	"sigs.k8s.io/kueue/pkg/util/slices"
+	utilslices "sigs.k8s.io/kueue/pkg/util/slices"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
@@ -2773,7 +2773,7 @@ func TestFairPreemptions(t *testing.T) {
 					},
 				},
 			), snapshotWorkingCopy)
-			gotTargets := sets.New(slices.Map(targets, func(t **Target) string {
+			gotTargets := sets.New(utilslices.Map(targets, func(t **Target) string {
 				return targetKeyReason(workload.Key((*t).WorkloadInfo.Obj), (*t).Reason)
 			})...)
 			if diff := cmp.Diff(tc.wantPreempted, gotTargets, cmpopts.EquateEmpty()); diff != "" {
@@ -2897,10 +2897,10 @@ func TestCandidatesOrdering(t *testing.T) {
 	_, log := utiltesting.ContextWithLog(t)
 	for _, tc := range cases {
 		features.SetFeatureGateDuringTest(t, features.AdmissionFairSharing, tc.admissionFairSharingEnabled)
-		sort.Slice(tc.candidates, func(i int, j int) bool {
-			return preemptioncommon.CandidatesOrdering(log, tc.admissionFairSharingEnabled, &tc.candidates[i], &tc.candidates[j], kueue.ClusterQueueReference(preemptorCq), now)
+		slices.SortFunc(tc.candidates, func(a, b workload.Info) int {
+			return preemptioncommon.CandidatesOrdering(log, tc.admissionFairSharingEnabled, &a, &b, kueue.ClusterQueueReference(preemptorCq), now)
 		})
-		got := slices.Map(tc.candidates, func(c *workload.Info) workload.Reference {
+		got := utilslices.Map(tc.candidates, func(c *workload.Info) workload.Reference {
 			return workload.Reference(c.Obj.Name)
 		})
 		if diff := cmp.Diff(tc.wantCandidates, got); diff != "" {
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/dispatcher_test.go b/test/integration/multikueue/dispatcher_test.go
index acc0b6f5d..2ad9a82a4 100644
--- a/test/integration/multikueue/dispatcher_test.go
+++ b/test/integration/multikueue/dispatcher_test.go
@@ -733,7 +733,5 @@ var _ = ginkgo.Describe("MultiKueueDispatcherAllAtOnce", ginkgo.Ordered, ginkgo.
 				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/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() {