Repository: kubernetes-sigs/kueue

Issue:
[flaky test] ManageJobsWithoutQueueName when manageJobsWithoutQueueName=true and LocalQueueDefaulting=false should suspend a job
/kind flake 

**What happened**:
failure https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/7899/pull-kueue-test-e2e-customconfigs-main/1993718563780694016
**What you expected to happen**:
no fail
**How to reproduce it (as minimally and precisely as possible)**:
ci
**Anything else we need to know?**:
```
End To End Custom Configs handling Suite: kindest/node:v1.34.0: [It] ManageJobsWithoutQueueName when manageJobsWithoutQueueName=true and LocalQueueDefaulting=false should suspend a job expand_less	51s
{Expected success, but got an error:
    <*errors.StatusError | 0xc000bfa1e0>: 
    conversion webhook for kueue.x-k8s.io/v1beta1, Kind=LocalQueue failed: Post "https://kueue-webhook-service.kueue-system.svc:443/convert?timeout=30s": EOF
    {
        ErrStatus: {
            TypeMeta: {Kind: "", APIVersion: ""},
            ListMeta: {
                SelfLink: "",
                ResourceVersion: "",
                Continue: "",
                RemainingItemCount: nil,
            },
            Status: "Failure",
            Message: "conversion webhook for kueue.x-k8s.io/v1beta1, Kind=LocalQueue failed: Post \"https://kueue-webhook-service.kueue-system.svc:443/convert?timeout=30s\": EOF",
            Reason: "",
            Details: nil,
            Code: 500,
        },
    } failed [FAILED] Expected success, but got an error:
    <*errors.StatusError | 0xc000bfa1e0>: 
    conversion webhook for kueue.x-k8s.io/v1beta1, Kind=LocalQueue failed: Post "https://kueue-webhook-service.kueue-system.svc:443/convert?timeout=30s": EOF
    {
        ErrStatus: {
            TypeMeta: {Kind: "", APIVersion: ""},
            ListMeta: {
                SelfLink: "",
                ResourceVersion: "",
                Continue: "",
                RemainingItemCount: nil,
            },
            Status: "Failure",
            Message: "conversion webhook for kueue.x-k8s.io/v1beta1, Kind=LocalQueue failed: Post \"https://kueue-webhook-service.kueue-system.svc:443/convert?timeout=30s\": EOF",
            Reason: "",
            Details: nil,
            Code: 500,
        },
    }
In [AfterEach] at: /home/prow/go/src/sigs.k8s.io/kueue/test/e2e/customconfigs/managejobswithoutqueuename_test.go:86 @ 11/26/25 16:41:21.303
}
```

Hints from issue comments:
@mimowo: I have analyzed the logs, and I think the reason is that while the Deployment is ready, the old pods may be still terminating, and sometimes receive still traffic, leading to errors. I think we should wait for `Status.TerminatingReplicas` to be 0.

To try to confirm this I looked at the kubelet logs around of the time of failure: `16:41:21`: https://storage.googleapis.com/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/7899/pull-kueue-test-e2e-customconfigs-main/1993718563780694016/artifacts/run-test-e2e-customconfigs-1.34.0/kind-worker2/kubelet.log

here we see the order of events:
- `16:41:21.009593` (kubelet)  the new Pod `kueue-controller-manager-bfc875747-v4pjr` is Ready
- `16:41:21.037487` (kubelet) the old Pod `kueue-controller-manager-579dfb7c48-8ddb8` receives delete
- `16:41:21.30` (test code) reports failure
- `16:41:22.043663` (kubelet)  the old Pod termination completes@mimowo: /assign

Solution patch:
diff --git a/pkg/cache/queue/manager.go b/pkg/cache/queue/manager.go
index 80156730e..265533524 100644
--- a/pkg/cache/queue/manager.go
+++ b/pkg/cache/queue/manager.go
@@ -445,6 +445,12 @@ func (m *Manager) AddOrUpdateWorkload(w *kueue.Workload, opts ...workload.InfoOp
 }
 
 func (m *Manager) AddOrUpdateWorkloadWithoutLock(w *kueue.Workload, opts ...workload.InfoOption) error {
+	if !workload.IsActive(w) {
+		return fmt.Errorf("workload %q is inactive and can't be added to a LocalQueue", w.Name)
+	}
+	if workload.HasQuotaReservation(w) {
+		return fmt.Errorf("workload %q already has quota reserved and can't be added to a LocalQueue", w.Name)
+	}
 	qKey := queue.KeyFromWorkload(w)
 	q := m.localQueues[qKey]
 	if q == nil {
diff --git a/pkg/cache/scheduler/tas_flavor_snapshot.go b/pkg/cache/scheduler/tas_flavor_snapshot.go
index 9d39cc312..e107eb315 100644
--- a/pkg/cache/scheduler/tas_flavor_snapshot.go
+++ b/pkg/cache/scheduler/tas_flavor_snapshot.go
@@ -31,6 +31,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
+	"k8s.io/component-helpers/scheduling/corev1/nodeaffinity"
 	"k8s.io/utils/ptr"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
@@ -98,13 +99,8 @@ type leafDomain struct {
 	// tasUsage represents the usage associated with TAS workloads.
 	tasUsage resources.Requests
 
-	// nodeTaints contains the list of taints for the node, only applies for
-	// lowest level of topology, if the lowest level is node
-	nodeTaints []corev1.Taint
-
-	// nodeLabels contains the list of labels on the node, only applies for
-	// lowest level of topology, if the lowest level is node
-	nodeLabels map[string]string
+	// node at the leaf, if the lowest level is a node
+	node *corev1.Node
 }
 
 type domainByID map[utiltas.TopologyDomainID]*domain
@@ -171,8 +167,7 @@ func (s *TASFlavorSnapshot) addNode(node corev1.Node) utiltas.TopologyDomainID {
 			},
 		}
 		if s.isLowestLevelNode() {
-			leafDomain.nodeTaints = slices.Clone(node.Spec.Taints)
-			leafDomain.nodeLabels = node.GetLabels()
+			leafDomain.node = &node
 		}
 		s.leaves[domainID] = &leafDomain
 	}
@@ -726,6 +721,18 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	} else {
 		selector = labels.Everything()
 	}
+
+	var affinitySelector *nodeaffinity.NodeSelector
+	if info.Affinity != nil && info.Affinity.NodeAffinity != nil {
+		if requiredAffinity := info.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution; requiredAffinity != nil {
+			var err error
+			affinitySelector, err = nodeaffinity.NewNodeSelector(requiredAffinity)
+			if err != nil {
+				return nil, fmt.Sprintf("invalid affinity node selectors: %s, reason: %s", requiredAffinity, err)
+			}
+		}
+	}
+
 	// phase 1 - determine the number of pods and slices which can fit in each topology domain
 	s.fillInCounts(
 		requests,
@@ -736,6 +743,7 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 		simulateEmpty,
 		append(podSetTolerations, s.tolerations...),
 		selector,
+		affinitySelector,
 		requiredReplacementDomain,
 	)
 
@@ -1306,6 +1314,7 @@ func (s *TASFlavorSnapshot) fillInCounts(
 	simulateEmpty bool,
 	tolerations []corev1.Toleration,
 	selector labels.Selector,
+	affinityNodeSelector *nodeaffinity.NodeSelector,
 	requiredReplacementDomain utiltas.TopologyDomainID) {
 	for _, domain := range s.domains {
 		// cleanup the state in case some remaining values are present from computing
@@ -1317,32 +1326,43 @@ func (s *TASFlavorSnapshot) fillInCounts(
 		domain.leaderState = 0
 	}
 	for _, leaf := range s.leaves {
-		// 1. Check Tolerations against Node Taints
-		taint, untolerated := corev1helpers.FindMatchingUntoleratedTaint(leaf.nodeTaints, tolerations, func(t *corev1.Taint) bool {
-			return t.Effect == corev1.TaintEffectNoSchedule || t.Effect == corev1.TaintEffectNoExecute
-		})
-		if untolerated {
-			s.log.V(3).Info("excluding node with untolerated taint", "domainID", leaf.id, "taint", taint)
-			continue
-		}
-		// 2. Check Node Labels against Compiled Selector
-		var nodeLabelSet labels.Set
-		if leaf.nodeLabels != nil {
-			nodeLabelSet = leaf.nodeLabels
+		// isLowestLevelNode() is necessary because we gather node level information only when
+		// node is the lowest level of the topology
+		if s.isLowestLevelNode() {
+			// 1. Check Tolerations against Node Taints
+			nodeTaints := leaf.node.Spec.Taints
+			taint, untolerated := corev1helpers.FindMatchingUntoleratedTaint(nodeTaints, tolerations, func(t *corev1.Taint) bool {
+				return t.Effect == corev1.TaintEffectNoSchedule || t.Effect == corev1.TaintEffectNoExecute
+			})
+			if untolerated {
+				s.log.V(3).Info("excluding node with untolerated taint", "domainID", leaf.id, "taint", taint)
+				continue
+			}
+
+			// 2. Check Node Labels against Compiled Selector
+			var nodeLabelSet labels.Set
+			if nodeLabels := leaf.node.GetLabels(); nodeLabels != nil {
+				nodeLabelSet = nodeLabels
+			}
+
+			if !selector.Matches(nodeLabelSet) {
+				s.log.V(3).Info("excluding node that doesn't match nodeSelectors", "domainID", leaf.id, "nodeLabels", nodeLabelSet)
+				continue
+			}
+
+			// 3. Check Node against Affinity Node Selector
+			if affinityNodeSelector != nil && !affinityNodeSelector.Match(leaf.node) {
+				s.log.V(3).Info("excluding node that doesn't match requiredDuringSchedulingIgnoredDuringExecution affinity", "domainID", leaf.id)
+				continue
+			}
 		}
 
-		// 3. While correcting the topologyAssignment with a failed node
+		// 4. While correcting the topologyAssignment with a failed node
 		// check if the leaf belongs to the required domain
 		if !belongsToRequiredDomain(leaf, requiredReplacementDomain) {
 			continue
 		}
 
-		// isLowestLevelNode() is necessary because we gather node level information only when
-		// node is the lowest level of the topology
-		if s.isLowestLevelNode() && !selector.Matches(nodeLabelSet) {
-			s.log.V(3).Info("excluding node that doesn't match nodeSelectors", "domainID", leaf.id, "nodeLabels", nodeLabelSet)
-			continue
-		}
 		remainingCapacity := leaf.freeCapacity.Clone()
 		if !simulateEmpty {
 			remainingCapacity.Sub(leaf.tasUsage)
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 596f02952..3f99f424e 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -1149,8 +1149,10 @@ func (h *resourceUpdatesHandler) queueReconcileForPending(ctx context.Context, q
 			continue
 		}
 
-		if err = h.r.queues.AddOrUpdateWorkload(wlCopy); err != nil {
-			log.V(2).Info("ignored an error for now", "error", err)
+		if workload.IsActive(wlCopy) && !workload.HasQuotaReservation(wlCopy) {
+			if err = h.r.queues.AddOrUpdateWorkload(wlCopy); err != nil {
+				log.V(2).Info("ignored an error for now", "error", err)
+			}
 		}
 	}
 }
diff --git a/pkg/podset/podset.go b/pkg/podset/podset.go
index d5e89ae70..5517777e6 100644
--- a/pkg/podset/podset.go
+++ b/pkg/podset/podset.go
@@ -46,6 +46,7 @@ type PodSetInfo struct {
 	Annotations     map[string]string
 	Labels          map[string]string
 	NodeSelector    map[string]string
+	Affinity        *corev1.Affinity
 	Tolerations     []corev1.Toleration
 	SchedulingGates []corev1.PodSchedulingGate
 }
@@ -107,6 +108,7 @@ func FromPodSet(ps *kueue.PodSet) PodSetInfo {
 		Annotations:     maps.Clone(ps.Template.Annotations),
 		Labels:          maps.Clone(ps.Template.Labels),
 		NodeSelector:    maps.Clone(ps.Template.Spec.NodeSelector),
+		Affinity:        ps.Template.Spec.Affinity,
 		Tolerations:     slices.Clone(ps.Template.Spec.Tolerations),
 		SchedulingGates: slices.Clone(ps.Template.Spec.SchedulingGates),
 	}
diff --git a/pkg/scheduler/flavorassigner/flavor_assigner_attempts.go b/pkg/scheduler/flavorassigner/flavor_assigner_attempts.go
new file mode 100644
index 000000000..064c95fbb
--- /dev/null
+++ b/pkg/scheduler/flavorassigner/flavor_assigner_attempts.go
@@ -0,0 +1,210 @@
+/*
+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 flavorassigner
+
+import (
+	"fmt"
+	"slices"
+	"sort"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
+	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
+)
+
+// FlavorAssignmentAttempt captures one attempted flavor and its worst-case outcome
+// across the requested resources.
+type FlavorAssignmentAttempt struct {
+	Flavor  kueue.ResourceFlavorReference
+	Mode    FlavorAssignmentMode
+	Borrow  int
+	Reasons []string
+}
+
+type FlavorAssignmentAttempts []FlavorAssignmentAttempt
+
+func newFlavorAssignmentAttempts(len int) FlavorAssignmentAttempts {
+	fa := make(FlavorAssignmentAttempts, 0, len)
+	return fa
+}
+
+func (fa *FlavorAssignmentAttempts) AddNoFitFlavorAttempt(flavor kueue.ResourceFlavorReference, status *Status) {
+	flavorAttempt := FlavorAssignmentAttempt{
+		Flavor: flavor,
+		Mode:   NoFit,
+	}
+	if status != nil {
+		flavorAttempt.Reasons = append(flavorAttempt.Reasons, status.reasons...)
+	}
+	*fa = append(*fa, flavorAttempt)
+}
+
+func (fa *FlavorAssignmentAttempts) AddRepresentativeModeFlavorAttempt(
+	flavor kueue.ResourceFlavorReference,
+	mode FlavorAssignmentMode,
+	maxBorrow int, reasons []string,
+) {
+	flavorAttempt := FlavorAssignmentAttempt{
+		Flavor: flavor,
+		Mode:   mode,
+		Borrow: maxBorrow,
+	}
+	if len(reasons) > 0 {
+		sort.Strings(reasons)
+		flavorAttempt.Reasons = append(flavorAttempt.Reasons, reasons...)
+	}
+	*fa = append(*fa, flavorAttempt)
+}
+
+func mergeFlavorAttemptsForResource(
+	dst map[kueue.ResourceFlavorReference]FlavorAssignmentAttempt,
+	src FlavorAssignmentAttempts,
+	resName corev1.ResourceName,
+	cq *schdcache.ClusterQueueSnapshot,
+) {
+	mergeFlavorAttempts(dst, src)
+
+	if len(src) > 0 {
+		rg := cq.RGByResource(resName)
+		rgFlavorSet := sets.New(rg.Flavors...)
+
+		present := make(map[kueue.ResourceFlavorReference]struct{}, len(src))
+		for _, c := range src {
+			present[c.Flavor] = struct{}{}
+		}
+
+		for flv, at := range dst {
+			if !rgFlavorSet.Has(flv) {
+				continue
+			}
+			if _, ok := present[flv]; !ok {
+				msg := fmt.Sprintf("flavor %s does not provide resource %s", flv, resName)
+				if at.Mode != NoFit {
+					at.Mode = NoFit
+				}
+				at.Reasons = mergeUnique(at.Reasons, []string{msg})
+				sort.Strings(at.Reasons)
+				dst[flv] = at
+			}
+		}
+	}
+}
+
+func mergeFlavorAttempts(dst map[kueue.ResourceFlavorReference]FlavorAssignmentAttempt, src FlavorAssignmentAttempts) {
+	for _, at := range src {
+		if existing, ok := dst[at.Flavor]; ok {
+			worst := existing.Mode
+			worst = min(at.Mode, worst)
+
+			maxBorrow := existing.Borrow
+			maxBorrow = max(at.Borrow, maxBorrow)
+
+			reasons := mergeUnique(existing.Reasons, at.Reasons)
+			sort.Strings(reasons)
+			dst[at.Flavor] = FlavorAssignmentAttempt{
+				Flavor:  at.Flavor,
+				Mode:    worst,
+				Borrow:  maxBorrow,
+				Reasons: reasons,
+			}
+			continue
+		}
+
+		cp := FlavorAssignmentAttempt{
+			Flavor:  at.Flavor,
+			Mode:    at.Mode,
+			Borrow:  at.Borrow,
+			Reasons: mergeUnique(nil, at.Reasons),
+		}
+		sort.Strings(cp.Reasons)
+		dst[at.Flavor] = cp
+	}
+}
+
+func finalizeFlavorAssignmentAttempts(consideredFlavors map[kueue.ResourceFlavorReference]FlavorAssignmentAttempt) []FlavorAssignmentAttempt {
+	finalConsidered := make([]FlavorAssignmentAttempt, 0, len(consideredFlavors))
+	for _, v := range consideredFlavors {
+		finalConsidered = append(finalConsidered, v)
+	}
+	slices.SortFunc(finalConsidered, func(a, b FlavorAssignmentAttempt) int {
+		return strings.Compare(string(a.Flavor), string(b.Flavor))
+	})
+	return finalConsidered
+}
+
+func mergeUnique(a, b []string) []string {
+	if len(a) == 0 && len(b) == 0 {
+		return nil
+	}
+	set := make(map[string]struct{}, len(a)+len(b))
+	for _, s := range a {
+		set[s] = struct{}{}
+	}
+	for _, s := range b {
+		set[s] = struct{}{}
+	}
+	out := make([]string, 0, len(set))
+	for s := range set {
+		out = append(out, s)
+	}
+	return out
+}
+
+func FormatFlavorAssignmentAttemptsForEvents(a Assignment) string {
+	if len(a.PodSets) == 0 {
+		return ""
+	}
+	var b strings.Builder
+	for _, ps := range a.PodSets {
+		if len(ps.FlavorAssignmentAttempts) == 0 {
+			continue
+		}
+		var attemptsBuilder strings.Builder
+		for _, att := range ps.FlavorAssignmentAttempts {
+			if att.Mode == Fit && att.Borrow == 0 {
+				continue
+			}
+			if attemptsBuilder.Len() > 0 {
+				attemptsBuilder.WriteString(", ")
+			}
+			attemptsBuilder.WriteString(string(att.Flavor))
+			attemptsBuilder.WriteString("(")
+			attemptsBuilder.WriteString(att.Mode.String())
+			if att.Borrow > 0 {
+				attemptsBuilder.WriteString(fmt.Sprintf(";borrow=%d", att.Borrow))
+			}
+			if len(att.Reasons) > 0 {
+				attemptsBuilder.WriteString(";")
+				attemptsBuilder.WriteString(strings.Join(att.Reasons, ";"))
+			}
+			attemptsBuilder.WriteString(")")
+		}
+		if attemptsBuilder.Len() == 0 {
+			continue
+		}
+		if b.Len() > 0 {
+			b.WriteString(" | ")
+		}
+		b.WriteString(string(ps.Name))
+		b.WriteString(": ")
+		b.WriteString(attemptsBuilder.String())
+	}
+	return b.String()
+}
diff --git a/pkg/scheduler/flavorassigner/flavorassigner.go b/pkg/scheduler/flavorassigner/flavorassigner.go
index 6714422c0..ca5d941ae 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner.go
@@ -242,6 +242,8 @@ type PodSetAssignment struct {
 
 	TopologyAssignment     *tas.TopologyAssignment
 	DelayedTopologyRequest *kueue.DelayedTopologyRequestState
+
+	FlavorAssignmentAttempts []FlavorAssignmentAttempt
 }
 
 // RepresentativeMode calculates the representative mode for this assignment as
@@ -586,6 +588,8 @@ func (a *FlavorAssigner) assignFlavors(log logr.Logger, counts []int32) Assignme
 			requests.Add(podset.podSet.Requests)
 		}
 
+		consideredFlavors := make(map[kueue.ResourceFlavorReference]FlavorAssignmentAttempt)
+
 		groupFlavors := make(ResourceAssignment)
 		for _, ips := range podSets {
 			for resName := range ips.podSetAssignment.Flavors {
@@ -600,8 +604,10 @@ func (a *FlavorAssigner) assignFlavors(log logr.Logger, counts []int32) Assignme
 				// No need to compute again.
 				continue
 			}
-			flavors, status := a.findFlavorForPodSets(log, psIDs, requests, resName, assignment.Usage.Quota)
-			if status.IsError() || len(flavors) == 0 {
+
+			flavors, status, considered := a.findFlavorForPodSets(log, psIDs, requests, resName, assignment.Usage.Quota)
+			mergeFlavorAttemptsForResource(consideredFlavors, considered, resName, a.cq)
+			if status.IsError() || (len(flavors) == 0 && len(requests) > 0) {
 				groupFlavors = nil
 				groupStatus = *status
 				break
@@ -611,12 +617,15 @@ func (a *FlavorAssigner) assignFlavors(log logr.Logger, counts []int32) Assignme
 				groupStatus.reasons = append(groupStatus.reasons, status.reasons...)
 			}
 		}
+
+		finalConsidered := finalizeFlavorAssignmentAttempts(consideredFlavors)
 		atLeastOnePodsAssignmentFailed := false
 		for _, podSet := range podSets {
 			podSetFlavors := utilmaps.FilterKeys(groupFlavors, slices.Collect(maps.Keys(podSet.podSet.Requests)))
 
 			podSet.podSetAssignment.Flavors = podSetFlavors
 			podSet.podSetAssignment.Status = groupStatus
+			podSet.podSetAssignment.FlavorAssignmentAttempts = finalConsidered
 
 			assignment.append(podSet.podSet.Requests, podSet.podSetAssignment)
 			if podSet.podSetAssignment.Status.IsError() || (len(podSet.podSet.Requests) > 0 && len(podSet.podSetAssignment.Flavors) == 0) {
@@ -679,7 +688,8 @@ func (a *Assignment) append(requests resources.Requests, psAssignment *PodSetAss
 
 // findFlavorForPodSets finds the flavor which can satisfy all the PodSet requests
 // for all resources in the same group as resName.
-// Returns the chosen flavor, along with the information about resources that need to be borrowed.
+// Returns the chosen flavor, along with the information about resources that need to be borrowed
+// and the list of flavors that were also considered.
 // If the flavor cannot be immediately assigned, it returns a status with
 // reasons or failure.
 func (a *FlavorAssigner) findFlavorForPodSets(
@@ -688,10 +698,10 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 	requests resources.Requests,
 	resName corev1.ResourceName,
 	assignmentUsage resources.FlavorResourceQuantities,
-) (ResourceAssignment, *Status) {
+) (ResourceAssignment, *Status, FlavorAssignmentAttempts) {
 	resourceGroup := a.cq.RGByResource(resName)
 	if resourceGroup == nil {
-		return nil, NewStatus(fmt.Sprintf("resource %s unavailable in ClusterQueue", resName))
+		return nil, NewStatus(fmt.Sprintf("resource %s unavailable in ClusterQueue", resName)), nil
 	}
 
 	status := NewStatus()
@@ -709,6 +719,7 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 
 	var bestAssignment ResourceAssignment
 	bestAssignmentMode := worstGranularMode()
+	consideredFlavors := newFlavorAssignmentAttempts(len(resourceGroup.Flavors))
 
 	// We will only check against the flavors' labels for the resource.
 	attemptedFlavorIdx := -1
@@ -717,16 +728,26 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 		attemptedFlavorIdx = idx
 		fName := resourceGroup.Flavors[idx]
 
-		if fit, err := a.checkFlavorForPodSets(log, fName, psIDs, podSets, selectors, status); !fit {
+		flavorStatus := NewStatus()
+
+		if fit, err := a.checkFlavorForPodSets(log, fName, psIDs, podSets, selectors, flavorStatus); !fit {
+			if flavorStatus != nil {
+				status.reasons = append(status.reasons, flavorStatus.reasons...)
+			}
+			consideredFlavors.AddNoFitFlavorAttempt(fName, flavorStatus)
 			if err != nil {
 				status.err = err
-				return nil, status
+				return nil, status, consideredFlavors
 			}
 			continue
 		}
+
 		assignments := make(ResourceAssignment, len(requests))
 		// Calculate representativeMode for this assignment as the worst mode among all requests.
 		representativeMode := bestGranularMode()
+		maxBorrow := 0
+		var flavorQuotaReasons []string
+
 		for rName, val := range requests {
 			// Ensure the same resource flavor is used for the workload slice as in the original admitted slice.
 			if features.Enabled(features.ElasticJobsViaWorkloadSlices) && a.replaceWorkloadSlice != nil {
@@ -737,7 +758,9 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 					if originalFlavor := preemptWorkloadRequests.Flavors[rName]; originalFlavor != fName {
 						// Flavor mismatch. Skip further checks for this resource.
 						representativeMode = worstGranularMode()
-						status.reasons = append(status.reasons, fmt.Sprintf("could not assign %s flavor since the original workload is assigned: %s", fName, originalFlavor))
+						msg := fmt.Sprintf("could not assign %s flavor since the original workload is assigned: %s", fName, originalFlavor)
+						status.reasons = append(status.reasons, msg)
+						flavorQuotaReasons = append(flavorQuotaReasons, msg)
 						break
 					}
 
@@ -752,8 +775,10 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 
 			preemptionMode, borrow, s := a.fitsResourceQuota(log, fr, assignmentUsage[fr], val, resQuota)
 			if s != nil {
+				flavorQuotaReasons = append(flavorQuotaReasons, s.reasons...)
 				status.reasons = append(status.reasons, s.reasons...)
 			}
+			maxBorrow = max(maxBorrow, borrow)
 			mode := granularMode{preemptionMode, borrowingLevel(borrow)}
 			if isPreferred(representativeMode, mode, a.cq.FlavorFungibility) {
 				representativeMode = mode
@@ -769,6 +794,9 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 				borrow: borrow,
 			}
 		}
+
+		consideredFlavors.AddRepresentativeModeFlavorAttempt(fName, representativeMode.preemptionMode.flavorAssignmentMode(), maxBorrow, flavorQuotaReasons)
+
 		if features.Enabled(features.FlavorFungibility) {
 			if !shouldTryNextFlavor(representativeMode, a.cq.FlavorFungibility) {
 				bestAssignment = assignments
@@ -784,7 +812,7 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 			bestAssignmentMode = representativeMode
 			if bestAssignmentMode.preemptionMode == fit {
 				// All the resources fit in the cohort, no need to check more flavors.
-				return bestAssignment, nil
+				return bestAssignment, nil, consideredFlavors
 			}
 		}
 	}
@@ -799,10 +827,10 @@ func (a *FlavorAssigner) findFlavorForPodSets(
 			}
 		}
 		if bestAssignmentMode.preemptionMode == fit {
-			return bestAssignment, nil
+			return bestAssignment, nil, consideredFlavors
 		}
 	}
-	return bestAssignment, status
+	return bestAssignment, status, consideredFlavors
 }
 
 func (a *FlavorAssigner) checkFlavorForPodSets(
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index c1f5ea92c..a762b4541 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -621,6 +621,7 @@ func (s *Scheduler) admit(ctx context.Context, e *entry, cq *schdcache.ClusterQu
 		PodSetAssignments: e.assignment.ToAPI(),
 	}
 
+	consideredStr := flavorassigner.FormatFlavorAssignmentAttemptsForEvents(e.assignment)
 	cacheWl, err := s.assumeWorkload(log, e, cq, admission)
 	if err != nil {
 		return err
@@ -638,7 +639,8 @@ func (s *Scheduler) admit(ctx context.Context, e *entry, cq *schdcache.ClusterQu
 		}, workload.WithLooseOnApply(), workload.WithRetryOnConflictForPatch())
 		if err == nil {
 			// Record metrics and events for quota reservation and admission
-			s.recordWorkloadAdmissionMetrics(newWorkload, e.Obj, admission)
+			s.recordWorkloadAdmissionMetrics(newWorkload, e.Obj, admission, consideredStr)
+
 			log.V(2).Info("Workload successfully admitted and assigned flavors", "assignments", admission.PodSetAssignments)
 			return
 		}
@@ -808,20 +810,25 @@ func (s *Scheduler) requeueAndUpdate(ctx context.Context, e entry) {
 }
 
 // recordWorkloadAdmissionMetrics records metrics and events for workload admission process
-func (s *Scheduler) recordWorkloadAdmissionMetrics(newWorkload, originalWorkload *kueue.Workload, admission *kueue.Admission) {
+func (s *Scheduler) recordWorkloadAdmissionMetrics(newWorkload, originalWorkload *kueue.Workload, admission *kueue.Admission, consideredFlavors string) {
 	waitTime := workload.QueuedWaitTime(newWorkload, s.clock)
 
-	s.recordQuotaReservationMetrics(newWorkload, originalWorkload, admission, waitTime)
+	s.recordQuotaReservationMetrics(newWorkload, originalWorkload, admission, waitTime, consideredFlavors)
 	s.recordWorkloadAdmissionEvents(newWorkload, originalWorkload, admission, waitTime)
 }
 
 // recordQuotaReservationMetrics records metrics and events for quota reservation
-func (s *Scheduler) recordQuotaReservationMetrics(newWorkload, originalWorkload *kueue.Workload, admission *kueue.Admission, waitTime time.Duration) {
+func (s *Scheduler) recordQuotaReservationMetrics(newWorkload, originalWorkload *kueue.Workload, admission *kueue.Admission, waitTime time.Duration, consideredFlavors string) {
 	if workload.HasQuotaReservation(originalWorkload) {
 		return
 	}
 
-	s.recorder.Eventf(newWorkload, corev1.EventTypeNormal, "QuotaReserved", "Quota reserved in ClusterQueue %v, wait time since queued was %.0fs", admission.ClusterQueue, waitTime.Seconds())
+	quotaReservedEventMessage := fmt.Sprintf("Quota reserved in ClusterQueue %v, wait time since queued was %.0fs", admission.ClusterQueue, waitTime.Seconds())
+	if consideredFlavors != "" {
+		quotaReservedEventMessage += fmt.Sprintf("; Flavors considered: %s", consideredFlavors)
+	}
+
+	s.recorder.Event(newWorkload, corev1.EventTypeNormal, "QuotaReserved", api.TruncateEventMessage(quotaReservedEventMessage))
 
 	priorityClassName := workload.PriorityClassName(newWorkload)
 	metrics.QuotaReservedWorkload(admission.ClusterQueue, priorityClassName, waitTime)



Test patch:
diff --git a/pkg/scheduler/flavorassigner/flavorassigner_test.go b/pkg/scheduler/flavorassigner/flavorassigner_test.go
index 154bfda7e..b72f09495 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner_test.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner_test.go
@@ -18,6 +18,7 @@ package flavorassigner
 
 import (
 	"errors"
+	"strings"
 	"testing"
 
 	"github.com/go-logr/logr"
@@ -49,6 +50,99 @@ var (
 	})
 )
 
+// flexAssertConsidered compares Considered Flavors
+// with the actual assignment, but with the rule:
+//
+//   - if actual has ANY Fit for that podset -> we don't require that all wanted
+//     flavors appear; it's OK if actual only returned the "winning" flavor;
+//   - if actual has NO Fit -> we require that all wanted flavors appear.
+func flexAssertConsidered(t *testing.T, want, got Assignment) {
+	wantByName := make(map[string]PodSetAssignment, len(want.PodSets))
+	for _, ps := range want.PodSets {
+		wantByName[string(ps.Name)] = ps
+	}
+	gotByName := make(map[string]PodSetAssignment, len(got.PodSets))
+	for _, ps := range got.PodSets {
+		gotByName[string(ps.Name)] = ps
+	}
+
+	for name, wantPS := range wantByName {
+		gotPS, ok := gotByName[name]
+		if !ok {
+			t.Errorf("podset %q: missing in actual assignment", name)
+			continue
+		}
+
+		if len(wantPS.FlavorAssignmentAttempts) == 0 {
+			continue
+		}
+
+		assertPodSetConsideredFlexible(t, name, wantPS.FlavorAssignmentAttempts, gotPS.FlavorAssignmentAttempts)
+	}
+}
+
+func assertPodSetConsideredFlexible(t *testing.T, podSetName string, want, got []FlavorAssignmentAttempt) {
+	wantByFlavor := make(map[kueue.ResourceFlavorReference]FlavorAssignmentAttempt, len(want))
+	for _, wa := range want {
+		wantByFlavor[wa.Flavor] = wa
+	}
+
+	gotByFlavor := make(map[kueue.ResourceFlavorReference]FlavorAssignmentAttempt, len(got))
+	hasFit := false
+	for _, ga := range got {
+		gotByFlavor[ga.Flavor] = ga
+		if ga.Mode == Fit {
+			hasFit = true
+		}
+	}
+
+	if hasFit {
+		for flavor, ga := range gotByFlavor {
+			wa, ok := wantByFlavor[flavor]
+			if !ok {
+				t.Errorf("podset %q: unexpected flavor %q in FlavorAssignmentAttempts (fit case), got=%#v", podSetName, flavor, ga)
+				continue
+			}
+
+			if ga.Mode == Fit && isDoesNotProvideOnly(wa) {
+				continue
+			}
+
+			if diff := cmp.Diff(wa, ga); diff != "" {
+				t.Errorf("podset %q: flavor %q mismatch (fit case) (-want +got):\n%s", podSetName, flavor, diff)
+			}
+		}
+
+		return
+	}
+
+	for flavor, wa := range wantByFlavor {
+		ga, ok := gotByFlavor[flavor]
+		if !ok {
+			t.Errorf("podset %q: expected flavor %q in FlavorAssignmentAttempts (no-fit case)", podSetName, flavor)
+			continue
+		}
+		if diff := cmp.Diff(wa, ga); diff != "" {
+			t.Errorf("podset %q: flavor %q mismatch (no-fit case) (-want +got):\n%s", podSetName, flavor, diff)
+		}
+	}
+}
+
+func isDoesNotProvideOnly(at FlavorAssignmentAttempt) bool {
+	if at.Mode != NoFit {
+		return false
+	}
+	if len(at.Reasons) == 0 {
+		return false
+	}
+	for _, r := range at.Reasons {
+		if !strings.Contains(r, "does not provide resource") {
+			return false
+		}
+	}
+	return true
+}
+
 type simulationResultForFlavor struct {
 	preemptionPossiblity     preemptioncommon.PreemptionPossibility
 	borrowingAfterSimulation int
@@ -139,7 +233,8 @@ func TestAssignFlavors(t *testing.T) {
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Mi"),
 						},
-						Count: 1,
+						Count:                    1,
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{{Flavor: "default", Mode: Fit}},
 					},
 				},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -177,6 +272,9 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "tainted", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -206,6 +304,9 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU: resource.MustParse("1"),
 					},
 					Count: 1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "taint_and_toleration", Mode: Fit},
+					},
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "taint_and_toleration", Resource: corev1.ResourceCPU}: 1_000,
@@ -238,7 +339,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU: resource.MustParse("2"),
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor default, 1 more needed"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "default",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor default, 1 more needed"},
+						},
+					},
+					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "default", Resource: corev1.ResourceCPU}: 2_000,
@@ -282,6 +390,20 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:    resource.MustParse("3"),
 						corev1.ResourceMemory: resource.MustParse("10Mi"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (3) > maximum capacity (2)"},
+						},
+						{Flavor: "two", Mode: Fit},
+						{Flavor: "b_one", Mode: Fit},
+						{
+							Flavor:  "b_two",
+							Mode:    NoFit,
+							Reasons: []string{"flavor b_two does not provide resource memory"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -322,6 +444,14 @@ func TestAssignFlavors(t *testing.T) {
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU: resource.MustParse("8"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (9) > maximum capacity (4)"},
+							},
+							{Flavor: "two", Mode: Fit},
+						},
 						Count: 4,
 					},
 					{
@@ -332,6 +462,14 @@ func TestAssignFlavors(t *testing.T) {
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU: resource.MustParse("1"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (9) > maximum capacity (4)"},
+							},
+							{Flavor: "two", Mode: Fit},
+						},
 						Count: 1,
 					}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -383,6 +521,14 @@ func TestAssignFlavors(t *testing.T) {
 							corev1.ResourceMemory: resource.MustParse("4"),
 							"example.com/gpu":     resource.MustParse("4"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"flavor one does not provide resource example.com/gpu"},
+							},
+							{Flavor: "two", Mode: Fit},
+						},
 						Count: 4,
 					},
 					{
@@ -395,6 +541,14 @@ func TestAssignFlavors(t *testing.T) {
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"flavor one does not provide resource example.com/gpu"},
+							},
+							{Flavor: "two", Mode: Fit},
+						},
 						Count: 1,
 					}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -441,6 +595,18 @@ func TestAssignFlavors(t *testing.T) {
 							"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (5) > maximum capacity (4)",
 							"insufficient quota for example.com/gpu in flavor two, previously considered podsets requests (0) + current podset request (4) > maximum capacity (0)",
 						),
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (5) > maximum capacity (4)"},
+							},
+							{
+								Flavor:  "two",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for example.com/gpu in flavor two, previously considered podsets requests (0) + current podset request (4) > maximum capacity (0)"},
+							},
+						},
 						Count: 4,
 					},
 					{
@@ -452,6 +618,18 @@ func TestAssignFlavors(t *testing.T) {
 							"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (5) > maximum capacity (4)",
 							"insufficient quota for example.com/gpu in flavor two, previously considered podsets requests (0) + current podset request (4) > maximum capacity (0)",
 						),
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (5) > maximum capacity (4)"},
+							},
+							{
+								Flavor:  "two",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for example.com/gpu in flavor two, previously considered podsets requests (0) + current podset request (4) > maximum capacity (0)"},
+							},
+						},
 						Count: 1,
 					}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
@@ -489,6 +667,13 @@ func TestAssignFlavors(t *testing.T) {
 					Status: *NewStatus(
 						"insufficient quota for memory in flavor b_one, previously considered podsets requests (0) + current podset request (10Mi) > maximum capacity (1Mi)",
 					),
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "b_one",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for memory in flavor b_one, previously considered podsets requests (0) + current podset request (10Mi) > maximum capacity (1Mi)"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
@@ -535,6 +720,15 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceMemory: resource.MustParse("10Mi"),
 						"example.com/gpu":     resource.MustParse("3"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "b_one", Mode: Fit},
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (3) > maximum capacity (2)"},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -604,6 +798,23 @@ func TestAssignFlavors(t *testing.T) {
 						"insufficient unused quota for memory in flavor two, 5Mi more needed",
 						"insufficient unused quota for example.com/gpu in flavor b_one, 1 more needed",
 					),
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "b_one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for example.com/gpu in flavor b_one, 1 more needed"},
+						},
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (3) > maximum capacity (2)"}},
+						{
+							Flavor:  "two",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for memory in flavor two, 5Mi more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Borrowing: 1,
@@ -643,6 +854,18 @@ func TestAssignFlavors(t *testing.T) {
 						"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (3) > maximum capacity (2)",
 						"insufficient quota for memory in flavor two, previously considered podsets requests (0) + current podset request (10Mi) > maximum capacity (5Mi)",
 					),
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (3) > maximum capacity (2)"},
+						},
+						{
+							Flavor:  "two",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for memory in flavor two, previously considered podsets requests (0) + current podset request (10Mi) > maximum capacity (5Mi)"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
@@ -673,6 +896,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("3"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "tainted",
+							Mode:    NoFit,
+							Reasons: []string{"untolerated taint {instance spot NoSchedule <nil>} in flavor tainted"},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -723,6 +954,13 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor: "one", Mode: NoFit,
+							Reasons: []string{"flavor one doesn't match node affinity"},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -777,6 +1015,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Mi"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"flavor one doesn't match node affinity"},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -839,6 +1085,9 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -887,6 +1136,17 @@ func TestAssignFlavors(t *testing.T) {
 						"flavor one doesn't match node affinity",
 						"flavor two doesn't match node affinity",
 					),
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one",
+							Mode:    NoFit,
+							Reasons: []string{"flavor one doesn't match node affinity"},
+						},
+						{
+							Flavor:  "two",
+							Mode:    NoFit,
+							Reasons: []string{"flavor two doesn't match node affinity"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
@@ -922,6 +1182,14 @@ func TestAssignFlavors(t *testing.T) {
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU: resource.MustParse("5"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (5) > maximum capacity (4)"},
+							},
+							{Flavor: "two", Mode: Fit},
+						},
 						Count: 1,
 					},
 					{
@@ -932,6 +1200,9 @@ func TestAssignFlavors(t *testing.T) {
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU: resource.MustParse("3"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{Flavor: "one", Mode: Fit},
+						},
 						Count: 1,
 					},
 				},
@@ -982,6 +1253,9 @@ func TestAssignFlavors(t *testing.T) {
 							corev1.ResourceCPU:    resource.MustParse("4"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{Flavor: "default", Mode: Fit, Borrow: 1},
+						},
 						Count: 1,
 					},
 					{
@@ -994,6 +1268,9 @@ func TestAssignFlavors(t *testing.T) {
 							corev1.ResourceCPU:    resource.MustParse("6"),
 							corev1.ResourceMemory: resource.MustParse("4Gi"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{Flavor: "default", Mode: Fit, Borrow: 1},
+						},
 						Count: 1,
 					},
 				},
@@ -1034,7 +1311,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU: resource.MustParse("2"),
 					},
 					Status: *NewStatus("insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (2) > maximum capacity (1)"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (2) > maximum capacity (1)"},
+						},
+					},
+					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
@@ -1080,7 +1364,15 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU: resource.MustParse("2"),
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 1 more needed"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 1 more needed"},
+						},
+					},
+					Count: 1,
 				}},
 				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1114,7 +1406,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU: resource.MustParse("2"),
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 1 more needed"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 1 more needed"},
+						},
+					},
+					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 2_000,
@@ -1161,7 +1460,15 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU: resource.MustParse("2"),
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 2 more needed"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 2 more needed"},
+						},
+					},
+					Count: 1,
 				}},
 				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1204,6 +1511,18 @@ func TestAssignFlavors(t *testing.T) {
 						"flavor one doesn't match node affinity",
 						"insufficient unused quota for cpu in flavor two, 1 more needed",
 					),
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"flavor one doesn't match node affinity"},
+						},
+						{
+							Flavor:  "two",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor two, 1 more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1254,6 +1573,18 @@ func TestAssignFlavors(t *testing.T) {
 							"insufficient unused quota for cpu in flavor one, 1 more needed",
 							"untolerated taint {instance spot NoSchedule <nil>} in flavor tainted",
 						),
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    Preempt,
+								Reasons: []string{"insufficient unused quota for cpu in flavor one, 1 more needed"},
+							},
+							{
+								Flavor:  "tainted",
+								Mode:    NoFit,
+								Reasons: []string{"untolerated taint {instance spot NoSchedule <nil>} in flavor tainted"},
+							},
+						},
 						Count: 1,
 					},
 					{
@@ -1268,6 +1599,18 @@ func TestAssignFlavors(t *testing.T) {
 							"insufficient quota for cpu in flavor one, previously considered podsets requests (2) + current podset request (10) > maximum capacity (4)",
 							"insufficient unused quota for cpu in flavor tainted, 3 more needed",
 						),
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (2) + current podset request (10) > maximum capacity (4)"},
+							},
+							{
+								Flavor:  "tainted",
+								Mode:    Preempt,
+								Reasons: []string{"insufficient unused quota for cpu in flavor tainted, 3 more needed"},
+							},
+						},
 						Count: 10,
 					},
 				},
@@ -1327,6 +1670,9 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("3"),
 						corev1.ResourcePods: resource.MustParse("3"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "default", Mode: Fit},
+					},
 					Count: 3,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1358,7 +1704,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourcePods: resource.MustParse("3"),
 					},
 					Status: *NewStatus("insufficient quota for pods in flavor default, previously considered podsets requests (0) + current podset request (3) > maximum capacity (2)"),
-					Count:  3,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "default",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for pods in flavor default, previously considered podsets requests (0) + current podset request (3) > maximum capacity (2)"},
+						},
+					},
+					Count: 3,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
@@ -1394,6 +1747,9 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("3"),
 						corev1.ResourcePods: resource.MustParse("3"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "default", Mode: Fit},
+					},
 					Count: 3,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1479,7 +1835,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 1 more needed"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 1 more needed"},
+						},
+					},
+					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: "cpu"}:  9_000,
@@ -1521,7 +1884,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 1 more needed"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 1 more needed"},
+						},
+					},
+					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: "cpu"}:  9_000,
@@ -1561,6 +1931,14 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("9"),
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 1 more needed"},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1612,6 +1990,13 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("9"),
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+						{
+							Flavor:  "two",
+							Reasons: []string{"insufficient quota for cpu in flavor two, previously considered podsets requests (0) + current podset request (9) > maximum capacity (1)"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1663,6 +2048,10 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("9"),
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1713,6 +2102,9 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("9"),
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1772,6 +2164,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1830,6 +2230,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1888,6 +2296,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1946,6 +2362,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -1997,6 +2421,10 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2048,6 +2476,10 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2094,6 +2526,13 @@ func TestAssignFlavors(t *testing.T) {
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU: resource.MustParse("12"),
 						},
+						FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+							{
+								Flavor:  "one",
+								Mode:    NoFit,
+								Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (12) > maximum capacity (11)"},
+							},
+						},
 						Count: 1,
 					},
 				},
@@ -2144,6 +2583,10 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("9"),
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2194,6 +2637,15 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:  resource.MustParse("9"),
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+
+						{
+							Flavor:  "two",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for cpu in flavor two, previously considered podsets requests (0) + current podset request (9) > maximum capacity (1)"},
+						},
+					},
 					Count: 1,
 				}},
 				Borrowing: 1,
@@ -2253,6 +2705,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("2"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 2 more needed"},
+						},
+						{Flavor: "two", Mode: Fit, Borrow: 1},
+					},
 					Count: 1,
 				}},
 				Borrowing: 1,
@@ -2311,6 +2771,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("2"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 2 more needed"},
+						},
+						{Flavor: "two", Mode: Fit, Borrow: 1},
+					},
 					Count: 1,
 				}},
 				Borrowing: 1,
@@ -2360,7 +2828,15 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourcePods: resource.MustParse("1"),
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 1 more needed"),
-					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 1 more needed"},
+						},
+					},
+					Count: 1,
 				}},
 				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2413,6 +2889,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2464,6 +2948,14 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2513,6 +3005,15 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2562,6 +3063,15 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("12"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Borrow:  1,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 10 more needed"},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2580,7 +3090,7 @@ func TestAssignFlavors(t *testing.T) {
 			clusterQueue: *utiltestingapi.MakeClusterQueue("test-clusterqueue").
 				ResourceGroup(
 					*utiltestingapi.MakeFlavorQuotas("one").
-						Resource(corev1.ResourceCPU, "2").
+						Resource(corev1.ResourceCPU, "3").
 						Resource(corev1.ResourceMemory, "1Gi").
 						Obj(),
 					*utiltestingapi.MakeFlavorQuotas("two").
@@ -2616,6 +3126,16 @@ func TestAssignFlavors(t *testing.T) {
 						corev1.ResourceCPU:    resource.MustParse("3"),
 						corev1.ResourceMemory: resource.MustParse("10Mi"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor: "one",
+							Mode:   NoFit,
+							Reasons: []string{
+								"could not assign one flavor since the original workload is assigned: two",
+							},
+						},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2672,6 +3192,18 @@ func TestAssignFlavors(t *testing.T) {
 						"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (1) > maximum capacity (500m)",
 						"could not assign two flavor since the original workload is assigned: one",
 					),
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    NoFit,
+							Reasons: []string{"insufficient quota for cpu in flavor one, previously considered podsets requests (0) + current podset request (1) > maximum capacity (500m)"},
+						},
+						{
+							Flavor:  "two",
+							Mode:    NoFit,
+							Reasons: []string{"could not assign two flavor since the original workload is assigned: one"},
+						},
+					},
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
@@ -2723,6 +3255,14 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 5 more needed"),
 					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 5 more needed"},
+						},
+						{Flavor: "two", Mode: Fit, Borrow: 1},
+					},
 				}},
 				Borrowing: 0,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2777,6 +3317,14 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 5 more needed"),
 					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "one",
+							Mode:    Preempt,
+							Reasons: []string{"insufficient unused quota for cpu in flavor one, 5 more needed"},
+						},
+						{Flavor: "two", Mode: Fit, Borrow: 1},
+					},
 				}},
 				Borrowing: 0,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -2826,6 +3374,10 @@ func TestAssignFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("10"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{Flavor: "one", Mode: Fit, Borrow: 1},
+						{Flavor: "two", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Borrowing: 0,
@@ -2906,10 +3458,14 @@ func TestAssignFlavors(t *testing.T) {
 
 			if diff := cmp.Diff(tc.wantAssignment, assignment,
 				cmpopts.EquateEmpty(),
-				cmpopts.IgnoreUnexported(Assignment{}, FlavorAssignment{}), statusComparer, cmpopts.IgnoreFields(Assignment{}, "LastState"),
+				cmpopts.IgnoreUnexported(Assignment{}, FlavorAssignment{}),
+				statusComparer, cmpopts.IgnoreFields(Assignment{}, "LastState"),
+				cmpopts.IgnoreFields(PodSetAssignment{}, "FlavorAssignmentAttempts"),
 			); diff != "" {
 				t.Errorf("Unexpected assignment (-want,+got):\n%s", diff)
 			}
+
+			flexAssertConsidered(t, tc.wantAssignment, assignment)
 		})
 	}
 }
@@ -3131,6 +3687,14 @@ func TestDeletedFlavors(t *testing.T) {
 					Requests: corev1.ResourceList{
 						corev1.ResourceCPU: resource.MustParse("3"),
 					},
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "deleted-flavor",
+							Mode:    NoFit,
+							Reasons: []string{"flavor deleted-flavor not found"},
+						},
+						{Flavor: "flavor", Mode: Fit},
+					},
 					Count: 1,
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
@@ -3158,6 +3722,13 @@ func TestDeletedFlavors(t *testing.T) {
 					},
 					Status: *NewStatus("flavor deleted-flavor not found"),
 					Count:  1,
+					FlavorAssignmentAttempts: []FlavorAssignmentAttempt{
+						{
+							Flavor:  "deleted-flavor",
+							Mode:    NoFit,
+							Reasons: []string{"flavor deleted-flavor not found"},
+						},
+					},
 				}},
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
diff --git a/pkg/scheduler/scheduler_tas_test.go b/pkg/scheduler/scheduler_tas_test.go
index fc6e9198a..01c2ed816 100644
--- a/pkg/scheduler/scheduler_tas_test.go
+++ b/pkg/scheduler/scheduler_tas_test.go
@@ -1478,10 +1478,11 @@ func TestScheduleForTAS(t *testing.T) {
 						Obj()).
 					Obj(),
 			},
-			eventCmpOpts: cmp.Options{eventIgnoreMessage},
 			wantEvents: []utiltesting.EventRecord{
-				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue tas-main, wait time since queued was 9223372037s; Flavors considered: one: default(NoFit;Flavor \"default\" does not support TopologyAwareScheduling)").Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-main, wait time since reservation was 0s").Obj(),
 			},
 		},
 		"workload which does not need TAS skips the TAS flavor": {
@@ -1512,10 +1513,11 @@ func TestScheduleForTAS(t *testing.T) {
 						Obj()).
 					Obj(),
 			},
-			eventCmpOpts: cmp.Options{eventIgnoreMessage},
 			wantEvents: []utiltesting.EventRecord{
-				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue tas-main, wait time since queued was 9223372037s; Flavors considered: one: tas-default(NoFit;Flavor \"tas-default\" supports only TopologyAwareScheduling)").Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-main, wait time since reservation was 0s").Obj(),
 			},
 		},
 		"workload with mixed PodSets (requiring TAS and not)": {
@@ -1559,10 +1561,11 @@ func TestScheduleForTAS(t *testing.T) {
 					).
 					Obj(),
 			},
-			eventCmpOpts: cmp.Options{eventIgnoreMessage},
 			wantEvents: []utiltesting.EventRecord{
-				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue tas-main, wait time since queued was 9223372037s; Flavors considered: launcher: tas-default(NoFit;Flavor \"tas-default\" supports only TopologyAwareScheduling)").Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-main, wait time since reservation was 0s").Obj(),
 			},
 		},
 		"workload required TAS gets scheduled": {
@@ -1669,10 +1672,11 @@ func TestScheduleForTAS(t *testing.T) {
 						Obj()).
 					Obj(),
 			},
-			eventCmpOpts: cmp.Options{eventIgnoreMessage},
 			wantEvents: []utiltesting.EventRecord{
-				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).
+					Message(`Quota reserved in ClusterQueue tas-main, wait time since queued was 9223372037s; Flavors considered: one: tas-default(NoFit;Flavor "tas-default" does not contain the requested level)`).Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-main, wait time since reservation was 0s").Obj(),
 			},
 		},
 		"workload does not get scheduled as it does not fit within the node capacity": {
@@ -2373,6 +2377,100 @@ func TestScheduleForTAS(t *testing.T) {
 					Message("Workload was evicted as there was no replacement for a failed node: x0").Obj(),
 			},
 		},
+		"does not admit workload when node does not match required affinity": {
+			nodes:           defaultSingleNode,
+			topologies:      []kueue.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueue},
+			workloads: []kueue.Workload{
+				*utiltestingapi.MakeWorkload("foo", "default").
+					Queue("tas-main").
+					PodSets(
+						*utiltestingapi.MakePodSet("one", 1).
+							PreferredTopologyRequest(corev1.LabelHostname).
+							RequiredDuringSchedulingIgnoredDuringExecution(
+								[]corev1.NodeSelectorTerm{
+									{
+										MatchExpressions: []corev1.NodeSelectorRequirement{
+											{
+												Key:      "unused-key",
+												Operator: corev1.NodeSelectorOpIn,
+												Values:   []string{"value"},
+											},
+										},
+									},
+								},
+							).
+							Request(corev1.ResourceCPU, "1").
+							Obj(),
+					).
+					Obj(),
+			},
+			wantInadmissibleLeft: map[kueue.ClusterQueueReference][]workload.Reference{
+				"tas-main": {"default/foo"},
+			},
+			eventCmpOpts: cmp.Options{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				utiltesting.MakeEventRecord("default", "foo", "Pending", corev1.EventTypeWarning).Obj(),
+			},
+		},
+		"admits workload when node matches required affinity": {
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("x1").
+					Label("tas-node", "true").
+					Label(corev1.LabelHostname, "x1").
+					Label("expected-label", "expected-value").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("1"),
+						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologies:      []kueue.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueue},
+			workloads: []kueue.Workload{
+				*utiltestingapi.MakeWorkload("foo", "default").
+					Queue("tas-main").
+					PodSets(
+						*utiltestingapi.MakePodSet("one", 1).
+							PreferredTopologyRequest(corev1.LabelHostname).
+							RequiredDuringSchedulingIgnoredDuringExecution(
+								[]corev1.NodeSelectorTerm{
+									{
+										MatchExpressions: []corev1.NodeSelectorRequirement{
+											{
+												Key:      "expected-label",
+												Operator: corev1.NodeSelectorOpIn,
+												Values:   []string{"expected-value"},
+											},
+										},
+									},
+								},
+							).
+							Request(corev1.ResourceCPU, "1").
+							Obj(),
+					).
+					Obj(),
+			},
+			wantNewAssignments: map[workload.Reference]kueue.Admission{
+				"default/foo": *utiltestingapi.MakeAdmission("tas-main").
+					PodSets(utiltestingapi.MakePodSetAssignment("one").Count(1).
+						Assignment(corev1.ResourceCPU, "tas-default", "1000m").
+						TopologyAssignment(utiltestingapi.MakeTopologyAssignment(utiltas.Levels(&defaultSingleLevelTopology)).
+							Domain(utiltestingapi.MakeTopologyDomainAssignment([]string{"x1"}, 1).Obj()).
+							Obj()).
+						Obj()).
+					Obj(),
+			},
+			eventCmpOpts: cmp.Options{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				utiltesting.MakeEventRecord("default", "foo", "QuotaReserved", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "foo", "Admitted", corev1.EventTypeNormal).Obj(),
+			},
+		},
 	}
 	for name, tc := range cases {
 		for _, enabled := range []bool{false, true} {
@@ -3618,10 +3716,11 @@ func TestScheduleForTASCohorts(t *testing.T) {
 						Obj()).
 					Obj(),
 			},
-			eventCmpOpts: cmp.Options{eventIgnoreMessage},
 			wantEvents: []utiltesting.EventRecord{
-				utiltesting.MakeEventRecord("default", "a1", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "a1", "Admitted", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "a1", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue tas-cq-a, wait time since queued was 9223372037s; Flavors considered: one: tas-default(Fit;borrow=1)").Obj(),
+				utiltesting.MakeEventRecord("default", "a1", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-cq-a, wait time since reservation was 0s").Obj(),
 			},
 		},
 		"reclaim within cohort; single borrowing workload gets preempted": {
@@ -4436,12 +4535,15 @@ func TestScheduleForTASCohorts(t *testing.T) {
 						Obj()).
 					Obj(),
 			},
-			eventCmpOpts: cmp.Options{eventIgnoreMessage},
 			wantEvents: []utiltesting.EventRecord{
-				utiltesting.MakeEventRecord("default", "b1", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "b1", "Admitted", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "a1", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "a1", "Admitted", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "b1", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue tas-cq-b, wait time since queued was 9223372037s").Obj(),
+				utiltesting.MakeEventRecord("default", "b1", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-cq-b, wait time since reservation was 0s").Obj(),
+				utiltesting.MakeEventRecord("default", "a1", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue tas-cq-a, wait time since queued was 9223372037s; Flavors considered: one: tas-default(Fit;borrow=1)").Obj(),
+				utiltesting.MakeEventRecord("default", "a1", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-cq-a, wait time since reservation was 0s").Obj(),
 			},
 		},
 		"two small workloads considered; both get scheduled on the same node": {
@@ -4754,11 +4856,13 @@ func TestScheduleForTASCohorts(t *testing.T) {
 						Obj()).
 					Obj(),
 			},
-			eventCmpOpts: cmp.Options{eventIgnoreMessage},
 			wantEvents: []utiltesting.EventRecord{
-				utiltesting.MakeEventRecord("default", "b1", "Pending", corev1.EventTypeWarning).Obj(),
-				utiltesting.MakeEventRecord("default", "a1", "QuotaReserved", corev1.EventTypeNormal).Obj(),
-				utiltesting.MakeEventRecord("default", "a1", "Admitted", corev1.EventTypeNormal).Obj(),
+				utiltesting.MakeEventRecord("default", "b1", "Pending", corev1.EventTypeWarning).
+					Message("Workload no longer fits after processing another workload").Obj(),
+				utiltesting.MakeEventRecord("default", "a1", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue tas-cq-a, wait time since queued was 9223372037s; Flavors considered: one: tas-default(Fit;borrow=1)").Obj(),
+				utiltesting.MakeEventRecord("default", "a1", "Admitted", corev1.EventTypeNormal).
+					Message("Admitted by ClusterQueue tas-cq-a, wait time since reservation was 0s").Obj(),
 			},
 		},
 		"two workloads considered; both overlapping in the initial flavor assignment": {
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 5a07e3bd6..8ac8ca023 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -7452,6 +7452,60 @@ func TestSchedule(t *testing.T) {
 					Obj(),
 			},
 		},
+		"pending admission check with nofit and fit flavors": {
+			workloads: []kueue.Workload{
+				*utiltestingapi.MakeWorkload("pending-check", "eng-beta").
+					Queue("main").
+					Request(corev1.ResourceCPU, "80").
+					AdmissionCheck(kueue.AdmissionCheckState{
+						Name:  "check",
+						State: kueue.CheckStatePending,
+					}).
+					Obj(),
+			},
+			wantAssignments: map[workload.Reference]kueue.Admission{
+				"eng-beta/pending-check": {
+					ClusterQueue: "eng-beta",
+					PodSetAssignments: []kueue.PodSetAssignment{
+						utiltestingapi.MakePodSetAssignment(kueue.DefaultPodSetName).
+							Assignment(corev1.ResourceCPU, "spot", "80").
+							Obj(),
+					},
+				},
+			},
+			wantWorkloads: []kueue.Workload{
+				*utiltestingapi.MakeWorkload("pending-check", "eng-beta").
+					Queue("main").
+					Request(corev1.ResourceCPU, "80").
+					AdmissionCheck(kueue.AdmissionCheckState{
+						Name:  "check",
+						State: kueue.CheckStatePending,
+					}).
+					Admission(
+						utiltestingapi.MakeAdmission("eng-beta").
+							PodSets(
+								utiltestingapi.MakePodSetAssignment(kueue.DefaultPodSetName).
+									Assignment(corev1.ResourceCPU, "spot", "80").
+									Obj(),
+							).
+							Obj(),
+					).
+					Condition(metav1.Condition{
+						Type:               kueue.WorkloadQuotaReserved,
+						Status:             metav1.ConditionTrue,
+						Reason:             kueue.WorkloadQuotaReserved,
+						Message:            "Quota reserved in ClusterQueue eng-beta",
+						LastTransitionTime: metav1.NewTime(now),
+					}).
+					Obj(),
+			},
+			wantEvents: []utiltesting.EventRecord{
+				utiltesting.MakeEventRecord("eng-beta", "pending-check", "QuotaReserved", corev1.EventTypeNormal).
+					Message("Quota reserved in ClusterQueue eng-beta, wait time since queued was 9223372037s; " +
+						"Flavors considered: main: on-demand(NoFit;insufficient quota for cpu in flavor on-demand, " +
+						"previously considered podsets requests (0) + current podset request (80) > maximum capacity (60)), spot(Fit;borrow=1)").Obj(),
+			},
+		},
 	}
 	for name, tc := range cases {
 		for _, enabled := range []bool{false, true} {
diff --git a/pkg/util/tas/tas_assignment_performance_test.go b/pkg/util/tas/tas_assignment_performance_test.go
new file mode 100644
index 000000000..80f84f1e0
--- /dev/null
+++ b/pkg/util/tas/tas_assignment_performance_test.go
@@ -0,0 +1,282 @@
+/*
+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 tas
+
+import (
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"math/rand/v2"
+	"sort"
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
+)
+
+const (
+	targetNodeCount = 40_000
+)
+
+// Generate n hex numbers of given length,
+// ensuring they're distinct (and otherwise quasi-random).
+// For the health of tests, this function behaves deterministically.
+func randomHexIDs(n, length int) []string {
+	chosen := map[string]bool{}
+	res := make([]string, n)
+	rnd := rand.NewChaCha8([32]byte{})
+	bytes := make([]byte, (length+1)/2)
+	for i := range n {
+		for {
+			var _, _ = rnd.Read(bytes)
+			id := hex.EncodeToString(bytes)[:length]
+			if !chosen[id] {
+				chosen[id] = true
+				res[i] = id
+				break
+			}
+		}
+	}
+	return res
+}
+
+func fixedID(length int) string {
+	// This is really whatever.
+	return strings.Repeat("a", length)
+}
+
+func consecutiveIPs(n int) []string {
+	res := make([]string, n)
+	for i := range n {
+		// Picking 100.10x.xxx.xxx (instead of traditional 10.x.xxx.xxx)
+		// to make the test scenario a bit more adverse.
+		res[i] = fmt.Sprintf("100.%d.%d.%d",
+			100+i/(1<<16),
+			i%(1<<16)/(1<<8),
+			i%(1<<8),
+		)
+	}
+	return res
+}
+
+// namingScheme generates a list of node names according to a specific recipe.
+//
+// The list should be "well-prefixing", in that namingScheme(n)[:m] should be
+// an "equally good representative" of the scheme as namingScheme(m).
+// (This convention helps test performance; it's used in "approxMaxNodesFor").
+// To achieve that, specific schemes should rotate node "properties"
+// (like node pool, region etc.) using "%" operator rather than in big fixed chunks.
+type namingScheme func(nodes int) []string
+
+type poolAndNodeBasedNamingConfig struct {
+	fixedPrefixAndSuffixLength int
+	pools                      int
+	nodeIDLength               int
+	poolIDLength               int
+}
+
+func poolAndNodeBasedNaming(config poolAndNodeBasedNamingConfig) namingScheme {
+	return func(nodes int) []string {
+		res := make([]string, nodes)
+		nodeIDs := randomHexIDs(nodes, config.nodeIDLength)
+		poolIDs := randomHexIDs(config.pools, config.poolIDLength)
+		for i := range nodes {
+			res[i] = fmt.Sprintf("%s-%s-%s-%s",
+				fixedID(config.fixedPrefixAndSuffixLength),
+				poolIDs[i%config.pools],
+				nodeIDs[i],
+				fixedID(config.fixedPrefixAndSuffixLength),
+			)
+		}
+		return res
+	}
+}
+
+type regionAndIPBasedNamingConfig struct {
+	fixedPrefixAndSuffixLength int
+	regions                    int
+	regionIDLength             int
+}
+
+func regionAndIPBasedNaming(config regionAndIPBasedNamingConfig) namingScheme {
+	return func(nodes int) []string {
+		res := make([]string, nodes)
+		nodeIPs := consecutiveIPs(nodes)
+		regionIDs := randomHexIDs(config.regions, config.regionIDLength)
+		for i := range nodes {
+			res[i] = fmt.Sprintf("%s-%s-%s-%s",
+				fixedID(config.fixedPrefixAndSuffixLength),
+				regionIDs[i%config.regions],
+				nodeIPs[i],
+				fixedID(config.fixedPrefixAndSuffixLength),
+			)
+		}
+		return res
+	}
+}
+
+type nodeBasedNamingConfig struct {
+	fixedPrefixAndSuffixLength int
+	nodeIDLength               int
+}
+
+func nodeBasedNaming(config nodeBasedNamingConfig) namingScheme {
+	return func(nodes int) []string {
+		res := make([]string, nodes)
+		nodeIDs := randomHexIDs(nodes, config.nodeIDLength)
+		for i := range nodes {
+			res[i] = fmt.Sprintf("%s-%s-%s",
+				fixedID(config.fixedPrefixAndSuffixLength),
+				nodeIDs[i],
+				fixedID(config.fixedPrefixAndSuffixLength),
+			)
+		}
+		return res
+	}
+}
+
+func internalSinglePodsOn(nodes []string) *TopologyAssignment {
+	res := &TopologyAssignment{
+		Levels:  []string{corev1.LabelHostname},
+		Domains: make([]TopologyDomainAssignment, len(nodes)),
+	}
+	for i, n := range nodes {
+		res.Domains[i] = TopologyDomainAssignment{
+			Values: []string{n},
+			Count:  1,
+		}
+	}
+	return res
+}
+
+func jsonStr(v any) string {
+	bytes, err := json.Marshal(v)
+	if err != nil {
+		panic(err)
+	}
+	return string(bytes)
+}
+
+func isTooLarge(ta *kueue.TopologyAssignment) bool {
+	return len(jsonStr(ta)) > 1_500_000
+}
+
+func approxMaxNodesFor(naming namingScheme) int {
+	step := 1_000 // We search with a reduced resolution, to speed up the test
+	ceiling := 300_000
+	nodeNames := naming(ceiling)
+	found := sort.Search(ceiling/step, func(n int) bool {
+		// Here we rely on "well-prefixing"; see the comment on "nodeNaming".
+		return isTooLarge(V1Beta2From(internalSinglePodsOn(nodeNames[:n*step])))
+	}) - 1
+	return found * step
+}
+
+type performanceTestCase struct {
+	name   string
+	naming namingScheme
+}
+
+var performanceTestCases = []performanceTestCase{
+	{
+		name: "pool-and-node-based naming (1000 node pools)",
+		naming: poolAndNodeBasedNaming(poolAndNodeBasedNamingConfig{
+			pools:        1000, // happens in practice, at least in GKE
+			nodeIDLength: 6,    // reached in AKS
+
+			// reachable in AKS (<pool-name>-<8-char-id>-vmss, then let <pool-name> have 8 chars)
+			poolIDLength: 22,
+
+			fixedPrefixAndSuffixLength: 20,
+		}),
+	},
+	{
+		name: "pool-and-node-based naming (10 node pools)",
+		naming: poolAndNodeBasedNaming(poolAndNodeBasedNamingConfig{
+			// Vendors tend to restrict "node pools" to 4k nodes or less,
+			// so, as we care about 40k+ nodes, it seems 10+ pools will always exist.
+			pools: 10,
+
+			nodeIDLength:               6,
+			poolIDLength:               22,
+			fixedPrefixAndSuffixLength: 20,
+		}),
+	},
+	{
+		name: "region-and-IP-based naming (100 regions)",
+		naming: regionAndIPBasedNaming(regionAndIPBasedNamingConfig{
+			regions: 100, // EKS has 70, leaving room for growth
+
+			// Reached in EKS ("ap-southeast-2") & ACK ("cn-zhangjiakou").
+			// GKE reaches even 23 ("northamerica-northeast2") but luckily its naming is not region-based.
+			regionIDLength: 14,
+
+			fixedPrefixAndSuffixLength: 20,
+		}),
+	},
+	{
+		name: "region-and-IP-based naming (1 region)",
+		naming: regionAndIPBasedNaming(regionAndIPBasedNamingConfig{
+			regions:                    1,
+			regionIDLength:             14,
+			fixedPrefixAndSuffixLength: 20,
+		}),
+	},
+	{
+		name: "node-only-based naming",
+		naming: nodeBasedNaming(nodeBasedNamingConfig{
+			nodeIDLength:               8, // reached in VKE
+			fixedPrefixAndSuffixLength: 20,
+		}),
+	},
+}
+
+func TestByteSizeLimit(t *testing.T) {
+	for _, tc := range performanceTestCases {
+		t.Run(tc.name, func(t *testing.T) {
+			nodesLimit := approxMaxNodesFor(tc.naming)
+			if nodesLimit < targetNodeCount {
+				t.Errorf("Nodes limit for naming %q is too low: got approx. %d, want >= %d", tc.name, nodesLimit, targetNodeCount)
+			} else {
+				t.Logf("Nodes limit for naming %q is approx. %d", tc.name, nodesLimit)
+			}
+		})
+	}
+}
+
+func BenchmarkV1Beta2From(b *testing.B) {
+	for _, tc := range performanceTestCases {
+		nodeNames := tc.naming(targetNodeCount)
+
+		// For our current strategy (just extract single common prefix & suffix),
+		// having node names sorted is an adverse scenario for benchmarking.
+		// (This is because longer common prefix & suffix will "hold" for longer).
+		// If we ever wish to test other approaches, we may want to also have
+		// a variant of this benchmark when node names get shuffled quasi-randomly.
+		sort.Strings(nodeNames)
+		ta := internalSinglePodsOn(nodeNames)
+
+		desc := fmt.Sprintf("Naming scheme %q, %d nodes", tc.name, targetNodeCount)
+		b.Run(desc, func(b *testing.B) {
+			for b.Loop() {
+				var _ = V1Beta2From(ta)
+			}
+		})
+	}
+}
diff --git a/pkg/util/testing/core.go b/pkg/util/testing/core.go
index 8c6a731cd..110e3bf8d 100644
--- a/pkg/util/testing/core.go
+++ b/pkg/util/testing/core.go
@@ -60,16 +60,23 @@ func CheckEventRecordedFor(ctx context.Context, k8sClient client.Client,
 		len(events.Items), eventReason, eventType, eventMessage, ref.Namespace)
 }
 
-// HasEventAppeared returns if an event has been emitted
-func HasEventAppeared(ctx context.Context, k8sClient client.Client, event corev1.Event) (bool, error) {
+// HasMatchingEventAppeared returns if an event has been emitted
+func HasMatchingEventAppeared(ctx context.Context, k8sClient client.Client, matcher func(*corev1.Event) bool) (bool, error) {
 	events := &corev1.EventList{}
 	if err := k8sClient.List(ctx, events, &client.ListOptions{}); err != nil {
 		return false, err
 	}
 	for _, item := range events.Items {
-		if item.Reason == event.Reason && item.Type == event.Type && item.Message == event.Message {
+		if matcher(&item) {
 			return true, nil
 		}
 	}
 	return false, nil
 }
+
+// HasEventAppeared returns if an event has been emitted
+func HasEventAppeared(ctx context.Context, k8sClient client.Client, event corev1.Event) (bool, error) {
+	return HasMatchingEventAppeared(ctx, k8sClient, func(item *corev1.Event) bool {
+		return item.Reason == event.Reason && item.Type == event.Type && item.Message == event.Message
+	})
+}
diff --git a/test/integration/singlecluster/controller/core/suite_test.go b/test/integration/singlecluster/controller/core/suite_test.go
index c15d63d03..f6d7fcb08 100644
--- a/test/integration/singlecluster/controller/core/suite_test.go
+++ b/test/integration/singlecluster/controller/core/suite_test.go
@@ -29,8 +29,10 @@ import (
 	config "sigs.k8s.io/kueue/apis/config/v1beta2"
 	qcache "sigs.k8s.io/kueue/pkg/cache/queue"
 	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
+	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/core"
 	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
+	"sigs.k8s.io/kueue/pkg/scheduler"
 	"sigs.k8s.io/kueue/pkg/webhooks"
 	"sigs.k8s.io/kueue/test/integration/framework"
 	"sigs.k8s.io/kueue/test/util"
@@ -63,12 +65,31 @@ var _ = ginkgo.AfterSuite(func() {
 	fwk.Teardown()
 })
 
+type managerSetupOpts struct {
+	runScheduler bool
+}
+
+type managerSetupOption func(*managerSetupOpts)
+
+func runScheduler(opts *managerSetupOpts) {
+	opts.runScheduler = true
+}
+
 func managerSetup(ctx context.Context, mgr manager.Manager) {
 	managerAndControllerSetup(nil)(ctx, mgr)
 }
 
-func managerAndControllerSetup(controllersCfg *config.Configuration) framework.ManagerSetup {
+func managerAndSchedulerSetup(ctx context.Context, mgr manager.Manager) {
+	managerAndControllerSetup(nil, runScheduler)(ctx, mgr)
+}
+
+func managerAndControllerSetup(controllersCfg *config.Configuration, options ...managerSetupOption) framework.ManagerSetup {
 	return func(ctx context.Context, mgr manager.Manager) {
+		var opts managerSetupOpts
+		for _, opt := range options {
+			opt(&opts)
+		}
+
 		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
@@ -88,5 +109,11 @@ func managerAndControllerSetup(controllersCfg *config.Configuration) framework.M
 
 		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, controllersCfg)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
+
+		if opts.runScheduler {
+			sched := scheduler.New(queues, cCache, mgr.GetClient(), mgr.GetEventRecorderFor(constants.AdmissionName))
+			err = sched.Start(ctx)
+			gomega.Expect(err).NotTo(gomega.HaveOccurred())
+		}
 	}
 }
diff --git a/test/integration/singlecluster/controller/core/workload_controller_test.go b/test/integration/singlecluster/controller/core/workload_controller_test.go
index 9ecc3525b..e5f859f74 100644
--- a/test/integration/singlecluster/controller/core/workload_controller_test.go
+++ b/test/integration/singlecluster/controller/core/workload_controller_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
+	nodev1 "k8s.io/api/node/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
@@ -612,6 +613,95 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
 	})
 })
 
+var _ = ginkgo.Describe("Workload controller interaction with scheduler", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		ns           *corev1.Namespace
+		clusterQueue *kueue.ClusterQueue
+		localQueue   *kueue.LocalQueue
+		wl           *kueue.Workload
+	)
+
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, managerAndSchedulerSetup)
+	})
+
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+
+	ginkgo.When("workload's runtime class is changed", func() {
+		var flavor *kueue.ResourceFlavor
+		var runtimeClass *nodev1.RuntimeClass
+
+		const runtimeClassName = "test-kueue-class"
+
+		ginkgo.BeforeEach(func() {
+			ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "core-workload-")
+			flavor = utiltestingapi.MakeResourceFlavor(flavorOnDemand).Obj()
+			util.MustCreate(ctx, k8sClient, flavor)
+			clusterQueue = utiltestingapi.MakeClusterQueue("cluster-queue").
+				ResourceGroup(*utiltestingapi.MakeFlavorQuotas(flavorOnDemand).Resource(corev1.ResourceCPU, "5", "5").Obj()).
+				Cohort("cohort").
+				Obj()
+			util.MustCreate(ctx, k8sClient, clusterQueue)
+			localQueue = utiltestingapi.MakeLocalQueue("queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+			util.MustCreate(ctx, k8sClient, localQueue)
+			runtimeClass = utiltesting.MakeRuntimeClass(runtimeClassName, "rc-handler-1").Obj()
+			util.MustCreate(ctx, k8sClient, runtimeClass)
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, flavor, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, runtimeClass, true)
+		})
+
+		ginkgo.It("should not temporarily admit an inactive workload", func() {
+			ginkgo.By("creating an inactive workload", func() {
+				wl = utiltestingapi.MakeWorkload("wl1", ns.Name).
+					Queue(kueue.LocalQueueName(localQueue.Name)).
+					Request(corev1.ResourceCPU, "1").
+					RuntimeClass(runtimeClassName).
+					Active(false).
+					Obj()
+				util.MustCreate(ctx, k8sClient, wl)
+
+				wlKey := client.ObjectKeyFromObject(wl)
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("changing the runtime class", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					runtimeClassKey := client.ObjectKeyFromObject(runtimeClass)
+					g.Expect(k8sClient.Get(ctx, runtimeClassKey, runtimeClass)).To(gomega.Succeed())
+					if runtimeClass.ObjectMeta.Annotations == nil {
+						runtimeClass.ObjectMeta.Annotations = map[string]string{}
+					}
+					runtimeClass.ObjectMeta.Annotations["foo"] = "bar"
+					g.Expect(k8sClient.Update(ctx, runtimeClass)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("checking no 'quota reseved' event appearing for the workload", func() {
+				gomega.Consistently(func(g gomega.Gomega) {
+					found, err := utiltesting.HasMatchingEventAppeared(ctx, k8sClient, func(e *corev1.Event) bool {
+						return e.Reason == "QuotaReserved" &&
+							e.Type == corev1.EventTypeNormal &&
+							e.InvolvedObject.Kind == "Workload" &&
+							e.InvolvedObject.Name == wl.Name &&
+							e.InvolvedObject.Namespace == wl.Namespace
+					})
+					g.Expect(err).NotTo(gomega.HaveOccurred())
+					g.Expect(found).To(gomega.BeTrue())
+				}, util.ShortTimeout, util.Interval).ShouldNot(gomega.Succeed())
+			})
+		})
+	})
+})
+
 var _ = ginkgo.Describe("Workload controller with resource retention", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
 	ginkgo.When("manager is setup with tiny retention period", func() {
 		var (
diff --git a/test/integration/singlecluster/scheduler/excluderesources/scheduler_test.go b/test/integration/singlecluster/scheduler/excluderesources/scheduler_test.go
new file mode 100644
index 000000000..9f8403681
--- /dev/null
+++ b/test/integration/singlecluster/scheduler/excluderesources/scheduler_test.go
@@ -0,0 +1,197 @@
+/*
+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 excluderesources
+
+import (
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	config "sigs.k8s.io/kueue/apis/config/v1beta2"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
+	utiltestingapi "sigs.k8s.io/kueue/pkg/util/testing/v1beta2"
+	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var _ = ginkgo.Describe("SchedulerWithExcludeResourcePrefixes", func() {
+	var (
+		defaultFlavor *kueue.ResourceFlavor
+		ns            *corev1.Namespace
+		cq            *kueue.ClusterQueue
+		lq            *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeEach(func() {
+		configuration := &config.Configuration{
+			Resources: &config.Resources{
+				ExcludeResourcePrefixes: []string{
+					"networking.example.com/",
+					"storage.example.com/",
+				},
+			},
+		}
+		fwk.StartManager(ctx, cfg, managerAndSchedulerSetup(configuration))
+
+		defaultFlavor = utiltestingapi.MakeResourceFlavor("default").Obj()
+		gomega.Expect(k8sClient.Create(ctx, defaultFlavor)).To(gomega.Succeed())
+
+		ns = &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				GenerateName: "exclude-",
+			},
+		}
+		gomega.Expect(k8sClient.Create(ctx, ns)).To(gomega.Succeed())
+
+		cq = utiltestingapi.MakeClusterQueue("test-cq").
+			ResourceGroup(
+				*utiltestingapi.MakeFlavorQuotas("default").
+					Resource(corev1.ResourceCPU, "10").
+					Resource(corev1.ResourceMemory, "10Gi").
+					Resource("networking.other.com/vpc", "1000").
+					Obj(),
+			).Obj()
+		gomega.Expect(k8sClient.Create(ctx, cq)).To(gomega.Succeed())
+
+		lq = utiltestingapi.MakeLocalQueue("test-lq", ns.Name).ClusterQueue(cq.Name).Obj()
+		gomega.Expect(k8sClient.Create(ctx, lq)).To(gomega.Succeed())
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, cq, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, defaultFlavor, true)
+	})
+
+	ginkgo.It("should admit workload ignoring excluded resource", func() {
+		ginkgo.By("Creating a workload that requests an excluded resource")
+		wl := utiltestingapi.MakeWorkload("test-wl", ns.Name).
+			Queue(kueue.LocalQueueName(lq.Name)).
+			Request(corev1.ResourceCPU, "1").
+			Request("networking.example.com/vpc", "100").
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, wl)).To(gomega.Succeed())
+
+		wlKey := types.NamespacedName{
+			Name:      wl.Name,
+			Namespace: ns.Name,
+		}
+
+		ginkgo.By("Verifying workload gets admitted")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+			g.Expect(workload.IsAdmitted(wl)).To(gomega.BeTrue())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("Verifying excluded resource is not in admission resource usage")
+		gomega.Expect(wl.Status.Admission).NotTo(gomega.BeNil())
+		gomega.Expect(wl.Status.Admission.PodSetAssignments).To(gomega.HaveLen(1))
+		resourceUsage := wl.Status.Admission.PodSetAssignments[0].ResourceUsage
+		gomega.Expect(resourceUsage).To(gomega.HaveKey(corev1.ResourceCPU))
+		gomega.Expect(resourceUsage).NotTo(gomega.HaveKey("networking.example.com/vpc"))
+	})
+
+	ginkgo.It("should ignore multiple excluded resource prefixes", func() {
+		ginkgo.By("Creating a workload that requests resources from multiple excluded prefixes")
+		wl := utiltestingapi.MakeWorkload("multi-exclude-wl", ns.Name).
+			Queue(kueue.LocalQueueName(lq.Name)).
+			Request(corev1.ResourceCPU, "2").
+			Request("networking.example.com/vpc", "10").
+			Request("storage.example.com/disk", "50").
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, wl)).To(gomega.Succeed())
+
+		wlKey := types.NamespacedName{
+			Name:      wl.Name,
+			Namespace: ns.Name,
+		}
+
+		ginkgo.By("Verifying workload admission")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+			g.Expect(workload.IsAdmitted(wl)).To(gomega.BeTrue())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("Verifying both excluded resources are not in resource usage")
+		gomega.Expect(wl.Status.Admission).NotTo(gomega.BeNil())
+		resourceUsage := wl.Status.Admission.PodSetAssignments[0].ResourceUsage
+		gomega.Expect(resourceUsage).To(gomega.HaveKey(corev1.ResourceCPU))
+		gomega.Expect(resourceUsage).NotTo(gomega.HaveKey("networking.example.com/vpc"))
+		gomega.Expect(resourceUsage).NotTo(gomega.HaveKey("storage.example.com/disk"))
+	})
+
+	ginkgo.It("should enforce quota for non-excluded resources", func() {
+		ginkgo.By("Creating a workload that consumes most of the CPU quota")
+		wl1 := utiltestingapi.MakeWorkload("quota-wl-1", ns.Name).
+			Queue(kueue.LocalQueueName(lq.Name)).
+			Request(corev1.ResourceCPU, "8").
+			Request("networking.example.com/vpc", "1000").
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, wl1)).To(gomega.Succeed())
+
+		wl1Key := types.NamespacedName{Name: wl1.Name, Namespace: ns.Name}
+
+		ginkgo.By("Waiting for first workload to be admitted")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, wl1Key, wl1)).To(gomega.Succeed())
+			g.Expect(workload.IsAdmitted(wl1)).To(gomega.BeTrue())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("Creating a second workload that would exceed CPU quota")
+		wl2 := utiltestingapi.MakeWorkload("quota-wl-2", ns.Name).
+			Queue(kueue.LocalQueueName(lq.Name)).
+			Request(corev1.ResourceCPU, "5").
+			Request("networking.example.com/vpc", "1").
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, wl2)).To(gomega.Succeed())
+
+		wl2Key := types.NamespacedName{Name: wl2.Name, Namespace: ns.Name}
+
+		ginkgo.By("Verifying second workload is not admitted due to CPU quota")
+		gomega.Consistently(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, wl2Key, wl2)).To(gomega.Succeed())
+			g.Expect(workload.IsAdmitted(wl2)).Should(gomega.BeFalse())
+		}, util.ConsistentDuration, util.ShortInterval).Should(gomega.Succeed())
+	})
+
+	ginkgo.It("should use exact prefix matching", func() {
+		ginkgo.By("Creating a workload with a resource that doesn't match the excluded prefix")
+		wl := utiltestingapi.MakeWorkload("prefix-test-wl", ns.Name).
+			Queue(kueue.LocalQueueName(lq.Name)).
+			Request(corev1.ResourceCPU, "1").
+			Request("networking.other.com/vpc", "100").
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, wl)).To(gomega.Succeed())
+
+		wlKey := types.NamespacedName{Name: wl.Name, Namespace: ns.Name}
+
+		ginkgo.By("Verifying workload is admitted")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+			g.Expect(workload.IsAdmitted(wl)).To(gomega.BeTrue())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("Verifying non-matching resource is NOT excluded (appears in resource usage)")
+		gomega.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+		gomega.Expect(wl.Status.Admission.PodSetAssignments).To(gomega.HaveLen(1))
+		resourceUsage := wl.Status.Admission.PodSetAssignments[0].ResourceUsage
+		gomega.Expect(resourceUsage).To(gomega.HaveKey(corev1.ResourceName("networking.other.com/vpc")))
+		gomega.Expect(resourceUsage).To(gomega.HaveKey(corev1.ResourceCPU))
+	})
+})
diff --git a/test/integration/singlecluster/scheduler/excluderesources/suite_test.go b/test/integration/singlecluster/scheduler/excluderesources/suite_test.go
new file mode 100644
index 000000000..d69309ac9
--- /dev/null
+++ b/test/integration/singlecluster/scheduler/excluderesources/suite_test.go
@@ -0,0 +1,107 @@
+/*
+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 excluderesources
+
+import (
+	"context"
+	"testing"
+
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	config "sigs.k8s.io/kueue/apis/config/v1beta2"
+	qcache "sigs.k8s.io/kueue/pkg/cache/queue"
+	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
+	"sigs.k8s.io/kueue/pkg/constants"
+	"sigs.k8s.io/kueue/pkg/controller/core"
+	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
+	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
+	"sigs.k8s.io/kueue/pkg/scheduler"
+	"sigs.k8s.io/kueue/pkg/webhooks"
+	"sigs.k8s.io/kueue/test/integration/framework"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var (
+	cfg       *rest.Config
+	k8sClient client.Client
+	ctx       context.Context
+	fwk       *framework.Framework
+)
+
+func TestSchedulerWithExcludeResourcePrefixes(t *testing.T) {
+	gomega.RegisterFailHandler(ginkgo.Fail)
+
+	ginkgo.RunSpecs(t,
+		"Scheduler with Exclude Resource Prefixes Suite",
+	)
+}
+
+var _ = ginkgo.BeforeSuite(func() {
+	fwk = &framework.Framework{
+		WebhookPath: util.WebhookPath,
+	}
+	cfg = fwk.Init()
+	ctx, k8sClient = fwk.SetupClient(cfg)
+})
+
+var _ = ginkgo.AfterSuite(func() {
+	fwk.Teardown()
+})
+
+func managerAndSchedulerSetup(configuration *config.Configuration) framework.ManagerSetup {
+	if configuration == nil {
+		configuration = &config.Configuration{}
+	}
+	return func(ctx context.Context, mgr manager.Manager) {
+		var queueOptions []qcache.Option
+		var cacheOptions []schdcache.Option
+
+		mgr.GetScheme().Default(configuration)
+
+		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		// Configure excludeResourcePrefixes
+		if configuration.Resources != nil && len(configuration.Resources.ExcludeResourcePrefixes) > 0 {
+			queueOptions = append(queueOptions,
+				qcache.WithExcludedResourcePrefixes(configuration.Resources.ExcludeResourcePrefixes))
+			cacheOptions = append(cacheOptions,
+				schdcache.WithExcludedResourcePrefixes(configuration.Resources.ExcludeResourcePrefixes))
+		}
+
+		cCache := schdcache.New(mgr.GetClient(), cacheOptions...)
+		queues := qcache.NewManager(mgr.GetClient(), cCache, queueOptions...)
+
+		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
+
+		failedWebhook, err := webhooks.Setup(mgr)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
+
+		err = workloadjob.SetupIndexes(ctx, mgr.GetFieldIndexer())
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		sched := scheduler.New(queues, cCache, mgr.GetClient(),
+			mgr.GetEventRecorderFor(constants.AdmissionName))
+		err = sched.Start(ctx)
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+	}
+}
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index a987800f9..760780031 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -2050,6 +2050,84 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 				})
 			})
 
+			ginkgo.It("should admit workload when node is edited to match the required affinity node selector terms", func() {
+				var (
+					wl1 *kueue.Workload
+				)
+				customLabelKey := "custom-label-key-1"
+				customLabelCorrectValue := "value-1"
+				customLabelWrongValue := "value-2"
+
+				ginkgo.By("creating a node missing the required label", func() {
+					nodes = []corev1.Node{
+						*testingnode.MakeNode("node-missing-label").
+							Label("node-group", "tas").
+							Label(utiltesting.DefaultBlockTopologyLevel, "b1").
+							Label(utiltesting.DefaultRackTopologyLevel, "r1").
+							Label(corev1.LabelHostname, "node-missing-label").
+							StatusAllocatable(corev1.ResourceList{
+								corev1.ResourceCPU:    resource.MustParse("1"),
+								corev1.ResourceMemory: resource.MustParse("1Gi"),
+								corev1.ResourcePods:   resource.MustParse("10"),
+							}).
+							Ready().
+							Obj(),
+					}
+					util.CreateNodesWithStatus(ctx, k8sClient, nodes)
+				})
+
+				ginkgo.By("creating a workload requiring the missing label via Affinity", func() {
+					wl1 = utiltestingapi.MakeWorkload("wl-needs-label", ns.Name).
+						Queue(kueue.LocalQueueName(localQueue.Name)).
+						PodSets(*utiltestingapi.MakePodSet("main", 1).
+							RequiredDuringSchedulingIgnoredDuringExecution([]corev1.NodeSelectorTerm{
+								{
+									MatchExpressions: []corev1.NodeSelectorRequirement{
+										{
+											Key:      customLabelKey,
+											Operator: corev1.NodeSelectorOpIn,
+											Values:   []string{customLabelCorrectValue},
+										},
+									},
+								},
+							}).
+							Request(corev1.ResourceCPU, "1").
+							Obj()).
+						Obj()
+					util.MustCreate(ctx, k8sClient, wl1)
+				})
+
+				ginkgo.By("verify the workload with missing label is inadmissible", func() {
+					util.ExpectWorkloadsToBePending(ctx, k8sClient, wl1)
+					util.ExpectPendingWorkloadsMetric(clusterQueue, 0, 1)
+				})
+
+				ginkgo.By("add a label to the node with correct key but wrong value", func() {
+					nodeToUpdate := &corev1.Node{}
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodes[0].Name}, nodeToUpdate)).Should(gomega.Succeed())
+					nodeToUpdate.Labels[customLabelKey] = customLabelWrongValue
+					gomega.Expect(k8sClient.Update(ctx, nodeToUpdate)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the workload with the correct label key but wrong value is inadmissible", func() {
+					util.ExpectWorkloadsToBePending(ctx, k8sClient, wl1)
+					util.ExpectPendingWorkloadsMetric(clusterQueue, 0, 1)
+				})
+
+				ginkgo.By("add the correct label to the node", func() {
+					nodeToUpdate := &corev1.Node{}
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodes[0].Name}, nodeToUpdate)).Should(gomega.Succeed())
+					nodeToUpdate.Labels[customLabelKey] = customLabelCorrectValue
+					gomega.Expect(k8sClient.Update(ctx, nodeToUpdate)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the workload gets admitted after label with correct key and value is added", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					util.ExpectAdmittedWorkloadsTotalMetric(clusterQueue, "", 1)
+					util.ExpectPendingWorkloadsMetric(clusterQueue, 0, 0)
+				})
+			})
+
 			ginkgo.It("should admit workload when node gets required label added", func() {
 				var (
 					wl1 *kueue.Workload