scheduler: add PreQueueingHint extension point to narrow pod evaluation on events

Add PreQueueingHintFn that plugins implement to return targeted pod keys
per event, narrowing which pods are evaluated on cluster events and avoiding
thundering herd on the scheduling queue. The DRA plugin implements this for
ResourceClaimTemplate-based workloads, falling back to evaluating all pods
for deallocation or errors. Per-plugin narrowing ensures each plugin\047s
QueueingHintFn is only called for pods that plugin\047s PreQueueingHintFn
identified.

Feature gate: SchedulerPreQueueingHints (beta, default=true)
KEP: kubernetes/enhancements#6132

hack/tools/instrumentation/documentation/documentation-list.yaml

@@ -6648,6 +6648,18 @@
   componentEndpoints:
   - component: kube-scheduler
     endpoint: /metrics
+- name: pre_queueing_hint_evaluations_total
+  subsystem: scheduler
+  help: Number of PreQueueingHint evaluations, labeled by plugin and result (all_pods
+    or narrowed).
+  type: Counter
+  stabilityLevel: ALPHA
+  labels:
+  - plugin
+  - result
+  componentEndpoints:
+  - component: kube-scheduler
+    endpoint: /metrics
 - name: preemption_evaluation_duration_seconds
   subsystem: scheduler
   help: Duration in seconds for identifying the target preemption victims.

pkg/features/kube_features.go

@@ -1047,6 +1047,12 @@ const (
 	// This allows to process potentially schedulable pods ASAP, eliminating a penalty effect of the backoff queue.
 	SchedulerPopFromBackoffQ featuregate.Feature = "SchedulerPopFromBackoffQ"
 
+	// owner: @geetasg
+	// kep: https://kep.k8s.io/6132
+	//
+	// Enables PreQueueingHint extension point to narrow pod evaluation on events.
+	SchedulerPreQueueingHints featuregate.Feature = "SchedulerPreQueueingHints"
+
 	// owner: @atosatto @yuanchen8911
 	// kep: http://kep.k8s.io/3902
 	//
@@ -2062,6 +2068,9 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 	SchedulerPopFromBackoffQ: {
 		{Version: version.MustParse("1.33"), Default: true, PreRelease: featuregate.Beta},
 	},
+	SchedulerPreQueueingHints: {
+		{Version: version.MustParse("1.37"), Default: true, PreRelease: featuregate.Beta},
+	},
 
 	SeparateTaintEvictionController: {
 		{Version: version.MustParse("1.29"), Default: true, PreRelease: featuregate.Beta},
@@ -2776,7 +2785,8 @@ var defaultKubernetesFeatureGateDependencies = map[featuregate.Feature][]feature
 
 	SchedulerAsyncPreemption: {},
 
-	SchedulerPopFromBackoffQ: {},
+	SchedulerPopFromBackoffQ:  {},
+	SchedulerPreQueueingHints: {},
 
 	SeparateTaintEvictionController: {},
 

pkg/scheduler/backend/queue/scheduling_queue.go

@@ -266,12 +266,15 @@ type PriorityQueue struct {
 	isCompositePodGroupEnabled bool
 	// isInPlacePodVerticalScalingSchedulerPreemptionEnabled indicates whether the feature gate InPlacePodVerticalScalingSchedulerPreemption is enabled.
 	isInPlacePodVerticalScalingSchedulerPreemptionEnabled bool
+	// isPreQueueingHintsEnabled indicates whether the SchedulerPreQueueingHints feature gate is enabled.
+	isPreQueueingHintsEnabled bool
 }
 
 // QueueingHintFunction is the wrapper of QueueingHintFn that has PluginName.
 type QueueingHintFunction struct {
-	PluginName     string
-	QueueingHintFn fwk.QueueingHintFn
+	PluginName        string
+	QueueingHintFn    fwk.QueueingHintFn
+	PreQueueingHintFn fwk.PreQueueingHintFn
 }
 
 // clusterEvent has the event and involved objects.
@@ -442,6 +445,7 @@ func NewPriorityQueue(
 	isOpportunisticBatchingEnabled := utilfeature.DefaultFeatureGate.Enabled(features.OpportunisticBatching)
 	isCompositePodGroupEnabled := utilfeature.DefaultFeatureGate.Enabled(features.CompositePodGroup)
 	isInPlacePodVerticalScalingSchedulerPreemptionEnabled := utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScalingSchedulerPreemption)
+	isPreQueueingHintsEnabled := utilfeature.DefaultFeatureGate.Enabled(features.SchedulerPreQueueingHints)
 	lessConverted := convertLessFn(lessFn)
 
 	backoffQ := newBackoffQueue(options.clock, options.podInitialBackoffDuration, options.podMaxBackoffDuration, lessConverted, isPopFromBackoffQEnabled)
@@ -466,6 +470,7 @@ func NewPriorityQueue(
 		isOpportunisticBatchingEnabled:    isOpportunisticBatchingEnabled,
 		isCompositePodGroupEnabled:        isCompositePodGroupEnabled,
 		isInPlacePodVerticalScalingSchedulerPreemptionEnabled: isInPlacePodVerticalScalingSchedulerPreemptionEnabled,
+		isPreQueueingHintsEnabled:                             isPreQueueingHintsEnabled,
 	}
 	var backoffQPopper backoffQPopper
 	if isPopFromBackoffQEnabled {
@@ -556,12 +561,12 @@ func (p *PriorityQueue) isPodGroupMember(pod *v1.Pod) bool {
 }
 
 // isEntityWorthRequeuing calls isPodWorthRequeuing for all pods belonging to the entity and returns the highest queueing strategy.
-func (p *PriorityQueue) isEntityWorthRequeuing(logger klog.Logger, entity framework.QueuedEntityInfo, event fwk.ClusterEvent, oldObj, newObj interface{}) queueingStrategy {
+func (p *PriorityQueue) isEntityWorthRequeuing(logger klog.Logger, entity framework.QueuedEntityInfo, event fwk.ClusterEvent, oldObj, newObj interface{}, hintKeys *preQueueingHintPodKeys) queueingStrategy {
 	// For pod groups, if any pod is worth requeuing, the whole group is worth it.
 	// But we should prioritize higher strategies.
 	bestStrategy := queueSkip
 	entity.ForEachPodInfo(func(pInfo *framework.QueuedPodInfo) bool {
-		strategy := p.isPodWorthRequeuing(logger, pInfo, event, oldObj, newObj)
+		strategy := p.isPodWorthRequeuing(logger, pInfo, event, oldObj, newObj, hintKeys)
 		if strategy > bestStrategy {
 			bestStrategy = strategy
 		}
@@ -579,7 +584,7 @@ func (p *PriorityQueue) isEntityWorthRequeuing(logger klog.Logger, entity framew
 // the scheduling queue is supposed to enqueue this Pod to activeQ, skipping backoffQ.
 // If any of pInfo.unschedulablePlugins return Queue,
 // the scheduling queue is supposed to enqueue this Pod to activeQ/backoffQ depending on the remaining backoff time of the Pod.
-func (p *PriorityQueue) isPodWorthRequeuing(logger klog.Logger, pInfo *framework.QueuedPodInfo, event fwk.ClusterEvent, oldObj, newObj interface{}) queueingStrategy {
+func (p *PriorityQueue) isPodWorthRequeuing(logger klog.Logger, pInfo *framework.QueuedPodInfo, event fwk.ClusterEvent, oldObj, newObj interface{}, hintKeys *preQueueingHintPodKeys) queueingStrategy {
 	rejectorPlugins := pInfo.UnschedulablePlugins.Union(pInfo.PendingPlugins)
 	if rejectorPlugins.Len() == 0 {
 		logger.V(6).Info("Worth requeuing because no failed plugins", "pod", klog.KObj(pInfo.Pod))
@@ -614,6 +619,7 @@ func (p *PriorityQueue) isPodWorthRequeuing(logger klog.Logger, pInfo *framework
 	}
 
 	pod := pInfo.Pod
+	podNN := types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}
 	queueStrategy := queueSkip
 	for eventToMatch, hintfns := range hintMap {
 		if !framework.MatchClusterEvents(eventToMatch, event) {
@@ -625,6 +631,15 @@ func (p *PriorityQueue) isPodWorthRequeuing(logger klog.Logger, pInfo *framework
 				// skip if it's not hintfn from rejectorPlugins.
 				continue
 			}
+			// Per-plugin PreQueueingHint narrowing: skip this plugin's QueueingHintFn
+			// if the pod wasn't identified by this plugin's PreQueueingHintFn.
+			if hintKeys != nil {
+				if pluginPods, ok := hintKeys.perPlugin[hintfn.PluginName]; ok {
+					if !pluginPods.Has(podNN) {
+						continue
+					}
+				}
+			}
 
 			start := time.Now()
 			hint, err := hintfn.QueueingHintFn(logger, pod, oldObj, newObj)
@@ -1033,7 +1048,7 @@ func (p *PriorityQueue) determineSchedulingHintForInFlightPod(logger klog.Logger
 	for _, e := range events {
 		logger.V(5).Info("Checking event for in-flight pod", "pod", klog.KObj(pInfo.Pod), "event", e.event.Label())
 
-		switch p.isPodWorthRequeuing(logger, pInfo, e.event, e.oldObj, e.newObj) {
+		switch p.isPodWorthRequeuing(logger, pInfo, e.event, e.oldObj, e.newObj, nil) {
 		case queueSkip:
 			continue
 		case queueImmediately:
@@ -1308,7 +1323,7 @@ func (p *PriorityQueue) flushUnschedulableEntitiesLeftover(logger klog.Logger) {
 	}
 
 	if len(entitiesToMove) > 0 {
-		p.moveEntitiesToActiveOrBackoffQueue(logger, entitiesToMove, framework.EventUnschedulableTimeout, nil, nil)
+		p.moveEntitiesToActiveOrBackoffQueue(logger, entitiesToMove, framework.EventUnschedulableTimeout, nil, nil, nil)
 	}
 }
 
@@ -1416,7 +1431,7 @@ func (p *PriorityQueue) Update(ctx context.Context, oldPod, newPod *v1.Pod) {
 			// where a pod re-queued to activeQ in requeueEntityWithQueueingStrategy is popped and processed quickly enough,
 			// so the write in its failure handler overwrites the pod object in PodInfo, which is accessed for logging.
 			entityRef := klog.KObj(entity)
-			hint := p.isEntityWorthRequeuing(logger, entity, evt, oldPod, newPod)
+			hint := p.isEntityWorthRequeuing(logger, entity, evt, oldPod, newPod, nil)
 			queue := p.requeueEntityWithQueueingStrategy(logger, entity, hint, evt.Label())
 			if queue != unschedulableQ {
 				logger.V(5).Info("Entity moved to an internal scheduling queue because the Pod is updated", "type", entity.Type(), "entity", entityRef, "pod", klog.KObj(newPod), "event", evt.Label(), "queue", queue)
@@ -1797,17 +1812,69 @@ func (p *PriorityQueue) moveAllToActiveOrBackoffQueue(logger klog.Logger, event
 		return
 	}
 
-	unschedulableEntities := make([]framework.QueuedEntityInfo, 0, len(p.unschedulableEntities.entityInfoMap))
+	entities, hintKeys := p.collectEntitiesToEvaluate(logger, event, oldObj, newObj, preCheck)
+	p.moveEntitiesToActiveOrBackoffQueue(logger, entities, event, oldObj, newObj, hintKeys)
+}
+
+// collectEntitiesToEvaluate determines which unschedulable entities should be evaluated
+// for requeueing based on the event. If PreQueueingHintFns narrow the set, only targeted
+// entities are returned; otherwise all entities passing preCheck are returned.
+//
+// NOTE: this function assumes lock has been acquired in caller
+func (p *PriorityQueue) collectEntitiesToEvaluate(logger klog.Logger, event fwk.ClusterEvent, oldObj, newObj interface{}, preCheck PreEnqueueCheck) ([]framework.QueuedEntityInfo, *preQueueingHintPodKeys) {
+	// Run PreQueueingHintFns to narrow the pod set.
+	hintKeys := p.runPreQueueingHintPlugins(logger, event, oldObj, newObj)
+
+	// TODO: Add PreQueueingHint support for CompositePodGroup entities.
+	// Until then, skip narrowing when CPG is enabled to avoid missing CPG entities.
+	if !p.isCompositePodGroupEnabled && hintKeys != nil {
+		logger.V(5).Info("PreQueueingHint narrowed pod set", "candidates", hintKeys.candidatePods.Len(), "total", len(p.unschedulableEntities.entityInfoMap))
+		// Directly look up entities for pods identified by PreQueueingHint.
+		entities := make([]framework.QueuedEntityInfo, 0, hintKeys.candidatePods.Len())
+		seen := make(map[string]bool)
+		for nn := range hintKeys.candidatePods {
+			entityKey := fwk.PodKey(nn.Namespace, nn.Name).String()
+			if entity, exists := p.unschedulableEntities.entityInfoMap[entityKey]; exists {
+				if preCheck == nil || preCheck(entity.(*framework.QueuedPodInfo).Pod) {
+					entities = append(entities, entity)
+				}
+				continue
+			}
+			// If not found as standalone pod, check if it belongs to a pod group
+			// via the pod's schedulingGroup field.
+			// TODO: Add support for CompositePodGroup (CPG) lookup.
+			if p.isGenericWorkloadEnabled {
+				pod, err := p.podLister.Pods(nn.Namespace).Get(nn.Name)
+				if err != nil {
+					logger.V(5).Info("Failed to get pod for PodGroup lookup", "pod", nn, "err", err)
+				} else if pod.Spec.SchedulingGroup != nil && pod.Spec.SchedulingGroup.PodGroupName != nil {
+					entityKey = fwk.PodGroupKey(nn.Namespace, *pod.Spec.SchedulingGroup.PodGroupName).String()
+					if entity, exists := p.unschedulableEntities.entityInfoMap[entityKey]; exists && !seen[entityKey] {
+						seen[entityKey] = true
+						// Only preCheck the specific pod identified by PreQueueingHint,
+						// not all members of the PodGroup.
+						if preCheck == nil || preCheck(pod) {
+							entities = append(entities, entity)
+						}
+					}
+				}
+			}
+		}
+		return entities, hintKeys
+	}
+
+	// No narrowing — evaluate all unschedulable entities.
+	entities := make([]framework.QueuedEntityInfo, 0, len(p.unschedulableEntities.entityInfoMap))
 	for _, entity := range p.unschedulableEntities.entityInfoMap {
 		entity.ForEachPodInfo(func(pInfo *framework.QueuedPodInfo) bool {
 			if preCheck == nil || preCheck(pInfo.Pod) {
-				unschedulableEntities = append(unschedulableEntities, entity)
+				entities = append(entities, entity)
 				return false
 			}
 			return true
 		})
 	}
-	p.moveEntitiesToActiveOrBackoffQueue(logger, unschedulableEntities, event, oldObj, newObj)
+	return entities, nil
 }
 
 // MoveAllToActiveOrBackoffQueue moves all pods from unschedulableEntities to activeQ or backoffQ.
@@ -1849,7 +1916,7 @@ func (p *PriorityQueue) requeueEntityWithQueueingStrategy(logger klog.Logger, en
 }
 
 // NOTE: this function assumes lock has been acquired in caller
-func (p *PriorityQueue) moveEntitiesToActiveOrBackoffQueue(logger klog.Logger, entityInfoList []framework.QueuedEntityInfo, event fwk.ClusterEvent, oldObj, newObj interface{}) {
+func (p *PriorityQueue) moveEntitiesToActiveOrBackoffQueue(logger klog.Logger, entityInfoList []framework.QueuedEntityInfo, event fwk.ClusterEvent, oldObj, newObj interface{}, hintKeys *preQueueingHintPodKeys) {
 	if !p.isEventOfInterest(logger, event) {
 		// No plugin is interested in this event.
 		return
@@ -1866,7 +1933,7 @@ func (p *PriorityQueue) moveEntitiesToActiveOrBackoffQueue(logger klog.Logger, e
 			continue
 		}
 
-		schedulingHint := p.isEntityWorthRequeuing(logger, entity, event, oldObj, newObj)
+		schedulingHint := p.isEntityWorthRequeuing(logger, entity, event, oldObj, newObj, hintKeys)
 		if schedulingHint == queueSkip {
 			// QueueingHintFn determined that this entity isn't worth putting to activeQ or backoffQ by this event.
 			logger.V(5).Info("Event is not making entity schedulable", "type", entity.Type(), "entity", entity, "event", event.Label())
@@ -2198,3 +2265,67 @@ func podGroupKey(podGroup *schedulingv1beta1.PodGroup) fwk.EntityKey {
 func compositePodGroupKey(cpg *schedulingv1alpha3.CompositePodGroup) fwk.EntityKey {
 	return fwk.CompositePodGroupKey(cpg.Namespace, cpg.Name)
 }
+
+// preQueueingHintPodKeys holds the result of running PreQueueingHintFns for an event.
+// candidatePods is the union of all pods that any plugin wants to re-evaluate.
+type preQueueingHintPodKeys struct {
+	candidatePods sets.Set[types.NamespacedName]
+	perPlugin     map[string]sets.Set[types.NamespacedName]
+}
+
+// runPreQueueingHintPlugins runs all registered PreQueueingHintFns for the given event
+// and returns a narrowed set of pod keys to evaluate, or nil if all pods should be evaluated.
+//
+// NOTE: this function assumes lock has been acquired in caller
+func (p *PriorityQueue) runPreQueueingHintPlugins(logger klog.Logger, event fwk.ClusterEvent, oldObj, newObj interface{}) *preQueueingHintPodKeys {
+	if !p.isPreQueueingHintsEnabled {
+		return nil
+	}
+	if oldObj == nil && newObj == nil {
+		return nil
+	}
+	if framework.ClusterEventIsWildCard(event) {
+		return nil
+	}
+	result := &preQueueingHintPodKeys{
+		perPlugin: make(map[string]sets.Set[types.NamespacedName]),
+	}
+	for _, hintMap := range p.queueingHintMap {
+		for eventToMatch, hintfns := range hintMap {
+			if !framework.MatchClusterEvents(eventToMatch, event) {
+				continue
+			}
+			for _, hintfn := range hintfns {
+				if hintfn.PreQueueingHintFn == nil {
+					continue
+				}
+				hintResult, err := hintfn.PreQueueingHintFn(logger, oldObj, newObj)
+				if err != nil {
+					logger.Error(err, "PreQueueingHintFn failed, falling back to evaluate all pods", "plugin", hintfn.PluginName)
+					return nil
+				}
+				if hintResult.AllPods {
+					// This plugin wants all pods evaluated; we cannot narrow.
+					p.metricsRecorder.ObserveCounterAsync(metrics.PreQueueingHintEvaluations, 1, hintfn.PluginName, "all_pods")
+					return nil
+				}
+				p.metricsRecorder.ObserveCounterAsync(metrics.PreQueueingHintEvaluations, 1, hintfn.PluginName, "narrowed")
+				keys := sets.New(hintResult.Pods...)
+				if existing, ok := result.perPlugin[hintfn.PluginName]; ok {
+					result.perPlugin[hintfn.PluginName] = existing.Union(keys)
+				} else {
+					result.perPlugin[hintfn.PluginName] = keys
+				}
+				if result.candidatePods == nil {
+					result.candidatePods = keys
+				} else {
+					result.candidatePods = result.candidatePods.Union(keys)
+				}
+			}
+		}
+	}
+	if result.candidatePods == nil {
+		return nil
+	}
+	return result
+}

pkg/scheduler/backend/queue/scheduling_queue_test.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -5827,7 +5828,7 @@ func Test_isPodWorthRequeuing(t *testing.T) {
 			count = 0 // reset count every time
 			logger, ctx := ktesting.NewTestContext(t)
 			q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(test.queueingHintMap))
-			actual := q.isPodWorthRequeuing(logger, test.podInfo, test.event, test.oldObj, test.newObj)
+			actual := q.isPodWorthRequeuing(logger, test.podInfo, test.event, test.oldObj, test.newObj, nil)
 			if actual != test.expected {
 				t.Errorf("isPodWorthRequeuing() = %v, want %v", actual, test.expected)
 			}
@@ -9513,3 +9514,800 @@ func TestPriorityQueue_DeferredPodGroupCompatibility(t *testing.T) {
 		})
 	}
 }
+func TestPriorityQueue_PreQueueingHint(t *testing.T) {
+	tests := []struct {
+		name           string
+		featureEnabled bool
+		pods           []*v1.Pod
+		hints          []*QueueingHintFunction
+		wantMoved      []string
+	}{
+		{
+			name:           "gate enabled, hint narrows to one pod",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+				st.MakePod().Name("pod3").Namespace("ns1").UID("3").Obj(),
+			},
+			hints: []*QueueingHintFunction{{
+				PluginName:     "foo",
+				QueueingHintFn: queueHintReturnQueue,
+				PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+					return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+				},
+			}},
+			wantMoved: []string{"pod1_ns1"},
+		},
+		{
+			name:           "gate enabled, hint AllPods evaluates all",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+			},
+			hints: []*QueueingHintFunction{{
+				PluginName:     "foo",
+				QueueingHintFn: queueHintReturnQueue,
+			}},
+			wantMoved: []string{"pod1_ns1", "pod2_ns1"},
+		},
+		{
+			name:           "gate enabled, one plugin AllPods means all pods evaluated",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+				st.MakePod().Name("pod3").Namespace("ns1").UID("3").Obj(),
+			},
+			hints: []*QueueingHintFunction{
+				{
+					PluginName:     "foo",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+					},
+				},
+				{
+					PluginName:     "bar",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{AllPods: true}, nil
+					},
+				},
+			},
+			wantMoved: []string{"pod1_ns1", "pod2_ns1", "pod3_ns1"},
+		},
+		{
+			name:           "gate disabled, hint ignored, all pods evaluated",
+			featureEnabled: false,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+				st.MakePod().Name("pod3").Namespace("ns1").UID("3").Obj(),
+			},
+			hints: []*QueueingHintFunction{{
+				PluginName:     "foo",
+				QueueingHintFn: queueHintReturnQueue,
+				PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+					return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+				},
+			}},
+			wantMoved: []string{"pod1_ns1", "pod2_ns1", "pod3_ns1"},
+		},
+		{
+			name:           "single plugin narrows to subset (2 of 3)",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+				st.MakePod().Name("pod3").Namespace("ns1").UID("3").Obj(),
+			},
+			hints: []*QueueingHintFunction{{
+				PluginName:     "foo",
+				QueueingHintFn: queueHintReturnQueue,
+				PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+					return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{
+						{Name: "pod1", Namespace: "ns1"},
+						{Name: "pod3", Namespace: "ns1"},
+					}}, nil
+				},
+			}},
+			wantMoved: []string{"pod1_ns1", "pod3_ns1"},
+		},
+		{
+			name:           "hint narrows but QueueingHint rejects",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+			},
+			hints: []*QueueingHintFunction{{
+				PluginName:     "foo",
+				QueueingHintFn: queueHintReturnSkip,
+				PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+					return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+				},
+			}},
+			wantMoved: []string{},
+		},
+		{
+			name:           "two plugins narrow to same pod",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+			},
+			hints: []*QueueingHintFunction{
+				{
+					PluginName:     "foo",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+					},
+				},
+				{
+					PluginName:     "bar",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+					},
+				},
+			},
+			wantMoved: []string{"pod1_ns1"},
+		},
+		{
+			name:           "two plugins narrow to same two pods",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+				st.MakePod().Name("pod3").Namespace("ns1").UID("3").Obj(),
+			},
+			hints: []*QueueingHintFunction{
+				{
+					PluginName:     "foo",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{
+							{Name: "pod1", Namespace: "ns1"},
+							{Name: "pod2", Namespace: "ns1"},
+						}}, nil
+					},
+				},
+				{
+					PluginName:     "bar",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{
+							{Name: "pod1", Namespace: "ns1"},
+							{Name: "pod2", Namespace: "ns1"},
+						}}, nil
+					},
+				},
+			},
+			wantMoved: []string{"pod1_ns1", "pod2_ns1"},
+		},
+		{
+			name:           "two plugins disjoint pods (union evaluated)",
+			featureEnabled: true,
+			pods: []*v1.Pod{
+				st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj(),
+				st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj(),
+				st.MakePod().Name("pod3").Namespace("ns1").UID("3").Obj(),
+			},
+			hints: []*QueueingHintFunction{
+				{
+					PluginName:     "foo",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+					},
+				},
+				{
+					PluginName:     "bar",
+					QueueingHintFn: queueHintReturnQueue,
+					PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+						return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod2", Namespace: "ns1"}}}, nil
+					},
+				},
+			},
+			wantMoved: []string{"pod1_ns1", "pod2_ns1"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerPreQueueingHints, tt.featureEnabled)
+			logger, ctx := ktesting.NewTestContext(t)
+
+			m := makeEmptyQueueingHintMapPerProfile()
+			m[""][nodeAdd] = tt.hints
+
+			q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+			for _, pod := range tt.pods {
+				q.Add(ctx, pod)
+				entity, err := q.Pop(logger)
+				if err != nil {
+					t.Fatalf("Pop failed: %v", err)
+				}
+				pInfo := entity.(*framework.QueuedPodInfo)
+				pInfo.UnschedulablePlugins = sets.New[string]("foo", "bar")
+				if err := q.AddUnschedulablePodIfNotPresent(logger, pInfo, q.SchedulingCycle()); err != nil {
+					t.Fatalf("AddUnschedulablePodIfNotPresent failed: %v", err)
+				}
+			}
+
+			q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), nil)
+
+			moved := sets.New[string]()
+			unschedPods := q.UnschedulablePods()
+			for _, pod := range tt.pods {
+				key := pod.Name + "_" + pod.Namespace
+				found := false
+				for _, up := range unschedPods {
+					if up.Name == pod.Name && up.Namespace == pod.Namespace {
+						found = true
+						break
+					}
+				}
+				if !found {
+					moved.Insert(key)
+				}
+			}
+
+			wantMoved := sets.New(tt.wantMoved...)
+			if !moved.Equal(wantMoved) {
+				t.Errorf("moved pods = %v, want %v", moved, wantMoved)
+			}
+		})
+	}
+}
+
+// TestPreQueueingHint_FlushRescue verifies that pods missed by a buggy PreQueueingHintFn
+// (not included in the narrowed set) are eventually rescued by periodic flush
+// after podMaxInUnschedulablePodsDuration expires.
+func TestPreQueueingHint_FlushRescue(t *testing.T) {
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerPreQueueingHints, true)
+	logger, ctx := ktesting.NewTestContext(t)
+
+	c := testingclock.NewFakeClock(time.Now())
+
+	// PreQueueingHintFn that only returns pod1 - pod2 will be "missed"
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "fakePlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithClock(c), WithQueueingHintMapPerProfile(m))
+
+	pod1 := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()
+	pod2 := st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj()
+
+	// Add both pods to unschedulable (simulating failed scheduling)
+	for _, pod := range []*v1.Pod{pod1, pod2} {
+		q.Add(ctx, pod)
+		entity, err := q.Pop(logger)
+		if err != nil {
+			t.Fatalf("Pop failed: %v", err)
+		}
+		pInfo := entity.(*framework.QueuedPodInfo)
+		pInfo.UnschedulablePlugins = sets.New[string]("fakePlugin")
+		if err := q.AddUnschedulablePodIfNotPresent(logger, pInfo, q.SchedulingCycle()); err != nil {
+			t.Fatalf("AddUnschedulablePodIfNotPresent failed: %v", err)
+		}
+	}
+
+	// Trigger event - PreQueueingHint only identifies pod1
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), nil)
+
+	// Verify pod1 was moved but pod2 remains unschedulable
+	unschedPods := q.UnschedulablePods()
+	pod2Found := false
+	for _, p := range unschedPods {
+		if p.Name == "pod2" && p.Namespace == "ns1" {
+			pod2Found = true
+		}
+		if p.Name == "pod1" && p.Namespace == "ns1" {
+			t.Errorf("pod1 should have been moved out of unschedulable, but was still there")
+		}
+	}
+	if !pod2Found {
+		t.Fatalf("pod2 should still be in unschedulable after event (missed by PreQueueingHint)")
+	}
+
+	// Pop pod1 from activeQ to confirm it was requeued
+	entity, err := q.Pop(logger)
+	if err != nil {
+		t.Fatalf("Pop failed after event: %v", err)
+	}
+	pInfo := entity.(*framework.QueuedPodInfo)
+	if pInfo.Pod.Name != "pod1" {
+		t.Errorf("expected pod1 to be popped, got %s", pInfo.Pod.Name)
+	}
+
+	// Advance clock past flush duration - pod2 should be rescued
+	c.Step(DefaultPodMaxInUnschedulablePodsDuration + time.Second)
+	q.flushUnschedulableEntitiesLeftover(logger)
+
+	// Verify pod2 is now in activeQ (rescued by flush)
+	unschedPods = q.UnschedulablePods()
+	for _, p := range unschedPods {
+		if p.Name == "pod2" && p.Namespace == "ns1" {
+			t.Errorf("pod2 should have been flushed to activeQ, but is still in unschedulable")
+		}
+	}
+
+	// Pop pod2 and verify it was rescued
+	entity, err = q.Pop(logger)
+	if err != nil {
+		t.Fatalf("Pop failed after flush: %v", err)
+	}
+	pInfo = entity.(*framework.QueuedPodInfo)
+	if pInfo.Pod.Name != "pod2" {
+		t.Errorf("expected pod2 to be popped after flush rescue, got %s", pInfo.Pod.Name)
+	}
+	if !pInfo.WasFlushedFromUnschedulable {
+		t.Errorf("expected WasFlushedFromUnschedulable to be true for rescued pod2")
+	}
+}
+
+// TestPreQueueingHint_Metrics verifies that the PreQueueingHintEvaluations metric
+// is incremented with the correct labels when PreQueueingHintFn is invoked.
+func TestPreQueueingHint_Metrics(t *testing.T) {
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerPreQueueingHints, true)
+	logger, ctx := ktesting.NewTestContext(t)
+
+	metrics.Register()
+	metrics.PreQueueingHintEvaluations.Reset()
+
+	// Plugin that narrows to one pod
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "testPlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+	pod := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()
+	q.Add(ctx, pod)
+	entity, _ := q.Pop(logger)
+	pInfo := entity.(*framework.QueuedPodInfo)
+	pInfo.UnschedulablePlugins = sets.New[string]("testPlugin")
+	if err := q.AddUnschedulablePodIfNotPresent(logger, pInfo, q.SchedulingCycle()); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), nil)
+	q.metricsRecorder.FlushMetrics()
+
+	// Verify metric was incremented with "narrowed" label
+	narrowedCount, err := testutil.GetCounterMetricValue(metrics.PreQueueingHintEvaluations.WithLabelValues("testPlugin", "narrowed"))
+	if err != nil {
+		t.Fatalf("Failed to get metric: %v", err)
+	}
+	if narrowedCount != 1 {
+		t.Errorf("Expected narrowed count = 1, got %v", narrowedCount)
+	}
+}
+
+func TestPreQueueingHint_PodGroupLookup(t *testing.T) {
+	// Verify that PreQueueingHint O(1) lookup correctly resolves a pod's
+	// entity key via schedulingGroup when the pod belongs to a PodGroup.
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.SchedulerPreQueueingHints: true,
+		features.GenericWorkload:           true,
+	})
+	logger, ctx := ktesting.NewTestContext(t)
+
+	pgName := "test-pg"
+	pod1 := st.MakePod().Name("pgpod1").Namespace("ns1").UID("pgpod1").Label("allow", "").PodGroupName(pgName).Obj()
+	pod2 := st.MakePod().Name("pgpod2").Namespace("ns1").UID("pgpod2").Label("allow", "").PodGroupName(pgName).Obj()
+	// A standalone pod not in the group.
+	pod3 := st.MakePod().Name("standalone").Namespace("ns1").UID("standalone").Label("allow", "").Obj()
+
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		// Hint returns pgpod1 (a pod group member).
+		return fwk.PreQueueingHintResult{
+			Pods: []types.NamespacedName{{Name: "pgpod1", Namespace: "ns1"}},
+		}, nil
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "testPlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	podGroup := st.MakePodGroup().Name(pgName).Namespace("ns1").Obj()
+
+	// Create queue with pod objects in the informer so podLister.Get works.
+	q := NewTestQueueWithObjects(ctx, newDefaultQueueSort(), []runtime.Object{pod1, pod2, pod3},
+		WithQueueingHintMapPerProfile(m))
+
+	// Set up pod group in unschedulable state.
+	setupInitialPodGroupState(t, ctx, q, []*v1.Pod{pod1, pod2}, stateUnschedulable, podGroup)
+
+	// Also mark UnschedulablePlugins on the entity so QueueingHint can move it.
+	pgLookup := newQueuedPodGroupInfoForLookup("ns1", "test-pg", fwk.PodGroupKeyType)
+	entity := q.unschedulableEntities.get(pgLookup)
+	if entity == nil {
+		t.Fatal("pod group not found in unschedulable entities")
+	}
+	entity.(*framework.QueuedPodGroupInfo).UnschedulablePlugins = sets.New[string]("testPlugin")
+
+	// Add standalone pod to unschedulable queue.
+	q.Add(ctx, pod3)
+	standaloneEntity := q.activeQ.delete(newQueuedPodInfoForLookup(pod3))
+	if standaloneEntity == nil {
+		t.Fatal("standalone pod not found in activeQ")
+	}
+	standaloneEntity.(*framework.QueuedPodInfo).UnschedulablePlugins = sets.New[string]("testPlugin")
+	q.unschedulableEntities.addOrUpdate(standaloneEntity, false, framework.EventUnscheduledPodAdd.Label(), nil)
+
+	// Fire event — hint returns only pgpod1. The O(1) lookup should:
+	// 1. Not find "pod/ns1/pgpod1" (it's stored as a pod group)
+	// 2. Look up pod via podLister, find schedulingGroup
+	// 3. Construct "podgroup/ns1/test-pg" and find the entity
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), nil)
+
+	// The pod group should have been moved out of unschedulable.
+	unschedEntity := q.unschedulableEntities.get(pgLookup)
+	if unschedEntity != nil {
+		t.Errorf("pod group should have been moved out of unschedulable queue")
+	}
+
+	// Standalone pod should still be in unschedulable (hint didn't include it).
+	standaloneLookup := newQueuedPodInfoForLookup(pod3)
+	if q.unschedulableEntities.get(standaloneLookup) == nil {
+		t.Errorf("standalone pod should remain in unschedulable queue")
+	}
+}
+
+func TestPreQueueingHint_NilObjects(t *testing.T) {
+	// Verify that getPreQueueingHintPodKeys returns nil when both oldObj and newObj are nil,
+	// since PreQueueingHintFn cannot narrow without an object to inspect.
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.SchedulerPreQueueingHints: true,
+	})
+	logger, ctx := ktesting.NewTestContext(t)
+
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		// Should never be called when both objects are nil.
+		t.Fatal("PreQueueingHintFn should not be called with nil objects")
+		return fwk.PreQueueingHintResult{}, nil
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "testPlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+	// Call with both nil — should return nil without calling PreQueueingHintFn.
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, nil, nil)
+}
+
+func TestPreQueueingHint_ErrorFallback(t *testing.T) {
+	// Verify that when PreQueueingHintFn returns an error, all pods are evaluated
+	// (fallback to AllPods behavior).
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.SchedulerPreQueueingHints: true,
+	})
+	logger, ctx := ktesting.NewTestContext(t)
+
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		return fwk.PreQueueingHintResult{}, fmt.Errorf("intentional error")
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "fakePlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+	pod1 := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()
+	pod2 := st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj()
+
+	// Add both pods to unschedulable.
+	for _, pod := range []*v1.Pod{pod1, pod2} {
+		q.Add(ctx, pod)
+		entity, err := q.Pop(logger)
+		if err != nil {
+			t.Fatalf("Pop failed: %v", err)
+		}
+		pInfo := entity.(*framework.QueuedPodInfo)
+		pInfo.UnschedulablePlugins = sets.New[string]("fakePlugin")
+		if err := q.AddUnschedulablePodIfNotPresent(logger, pInfo, q.SchedulingCycle()); err != nil {
+			t.Fatalf("AddUnschedulablePodIfNotPresent failed: %v", err)
+		}
+	}
+
+	// Trigger event - PreQueueingHint returns error, so all pods should be evaluated.
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), nil)
+
+	// Both pods should be moved out of unschedulable (fallback to AllPods).
+	unschedPods := q.UnschedulablePods()
+	if len(unschedPods) != 0 {
+		t.Errorf("expected 0 pods in unschedulable after error fallback (all should be moved), got %d", len(unschedPods))
+	}
+}
+
+func TestPreQueueingHint_PodGroupPreCheck(t *testing.T) {
+	// Verify that preCheck is applied to PodGroup entities in the narrowed path.
+	// When preCheck rejects all member pods, the PodGroup entity should not be moved.
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.SchedulerPreQueueingHints: true,
+		features.GenericWorkload:           true,
+	})
+	logger, ctx := ktesting.NewTestContext(t)
+
+	pgName := "test-pg"
+	pod1 := st.MakePod().Name("pgpod1").Namespace("ns1").UID("pgpod1").Label("block", "").PodGroupName(pgName).Obj()
+	pod2 := st.MakePod().Name("pgpod2").Namespace("ns1").UID("pgpod2").Label("block", "").PodGroupName(pgName).Obj()
+	podGroup := &schedulingv1beta1.PodGroup{
+		ObjectMeta: metav1.ObjectMeta{Name: pgName, Namespace: "ns1", UID: "pg-uid"},
+		Spec:       schedulingv1beta1.PodGroupSpec{},
+	}
+
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		return fwk.PreQueueingHintResult{
+			Pods: []types.NamespacedName{{Name: "pgpod1", Namespace: "ns1"}},
+		}, nil
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "testPlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+	// Set up pod group in unschedulable state.
+	setupInitialPodGroupState(t, ctx, q, []*v1.Pod{pod1, pod2}, stateUnschedulable, podGroup)
+
+	pgLookup := newQueuedPodGroupInfoForLookup("ns1", pgName, fwk.PodGroupKeyType)
+	entity := q.unschedulableEntities.get(pgLookup)
+	if entity == nil {
+		t.Fatal("pod group not found in unschedulable entities")
+	}
+	entity.(*framework.QueuedPodGroupInfo).UnschedulablePlugins = sets.New[string]("testPlugin")
+
+	// preCheck rejects pods with label "block"
+	preCheck := func(pod *v1.Pod) bool {
+		_, hasBlock := pod.Labels["block"]
+		return !hasBlock
+	}
+
+	// Trigger event with preCheck that blocks all PodGroup member pods.
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), preCheck)
+
+	// PodGroup should remain in unschedulable because preCheck rejected all members.
+	if q.unschedulableEntities.get(pgLookup) == nil {
+		t.Errorf("pod group should remain in unschedulable when preCheck rejects all members")
+	}
+}
+
+func TestPreQueueingHint_CPGDisablesNarrowing(t *testing.T) {
+	// When CompositePodGroup feature gate is enabled, PreQueueingHint
+	// narrowing is disabled to avoid missing CPG entities.
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.SchedulerPreQueueingHints:       true,
+		features.GenericWorkload:                 true,
+		features.CompositePodGroup:               true,
+		features.TopologyAwareWorkloadScheduling: true,
+	})
+	logger, ctx := ktesting.NewTestContext(t)
+
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		// This should NOT be called when CPG is enabled.
+		// With CPG guard at collectEntitiesToEvaluate, hint fn is called but result ignored.
+		return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "testPlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+	pod1 := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()
+	pod2 := st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj()
+
+	for _, pod := range []*v1.Pod{pod1, pod2} {
+		q.Add(ctx, pod)
+		entity, err := q.Pop(logger)
+		if err != nil {
+			t.Fatalf("Pop failed: %v", err)
+		}
+		pInfo := entity.(*framework.QueuedPodInfo)
+		pInfo.UnschedulablePlugins = sets.New[string]("testPlugin")
+		if err := q.AddUnschedulablePodIfNotPresent(logger, pInfo, q.SchedulingCycle()); err != nil {
+			t.Fatalf("AddUnschedulablePodIfNotPresent failed: %v", err)
+		}
+	}
+
+	// Trigger event — with CPG enabled, all pods should be evaluated (no narrowing).
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), nil)
+
+	// Both pods should be moved (narrowing disabled).
+	unschedPods := q.UnschedulablePods()
+	if len(unschedPods) != 0 {
+		t.Errorf("expected 0 pods in unschedulable (all should be moved), got %d", len(unschedPods))
+	}
+}
+
+func TestPreQueueingHint_WildcardSkipsNarrowing(t *testing.T) {
+	// When a wildcard event fires, PreQueueingHint narrowing is skipped
+	// and all pods are evaluated.
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.SchedulerPreQueueingHints: true,
+	})
+	logger, ctx := ktesting.NewTestContext(t)
+
+	preQueueingHintFn := func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+		// Should NOT be called for wildcard events.
+		t.Fatal("PreQueueingHintFn should not be called for wildcard events")
+		return fwk.PreQueueingHintResult{}, nil
+	}
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName:        "testPlugin",
+			QueueingHintFn:    queueHintReturnQueue,
+			PreQueueingHintFn: preQueueingHintFn,
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+	pod1 := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()
+	pod2 := st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj()
+
+	for _, pod := range []*v1.Pod{pod1, pod2} {
+		q.Add(ctx, pod)
+		entity, err := q.Pop(logger)
+		if err != nil {
+			t.Fatalf("Pop failed: %v", err)
+		}
+		pInfo := entity.(*framework.QueuedPodInfo)
+		pInfo.UnschedulablePlugins = sets.New[string]("testPlugin")
+		if err := q.AddUnschedulablePodIfNotPresent(logger, pInfo, q.SchedulingCycle()); err != nil {
+			t.Fatalf("AddUnschedulablePodIfNotPresent failed: %v", err)
+		}
+	}
+
+	// Fire wildcard event — narrowing should be skipped, all pods evaluated.
+	q.MoveAllToActiveOrBackoffQueue(logger, framework.EventUnschedulableTimeout, nil, nil, nil)
+
+	// Both pods should be moved.
+	unschedPods := q.UnschedulablePods()
+	if len(unschedPods) != 0 {
+		t.Errorf("expected 0 pods in unschedulable after wildcard event, got %d", len(unschedPods))
+	}
+}
+func TestPreQueueingHint_PerPluginNarrowing(t *testing.T) {
+	featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
+		features.SchedulerPreQueueingHints: true,
+	})
+	logger, ctx := ktesting.NewTestContext(t)
+
+	var pluginAQueueingHintCalled []string
+	var pluginBQueueingHintCalled []string
+
+	m := makeEmptyQueueingHintMapPerProfile()
+	m[""][nodeAdd] = []*QueueingHintFunction{
+		{
+			PluginName: "pluginA",
+			QueueingHintFn: func(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (fwk.QueueingHint, error) {
+				pluginAQueueingHintCalled = append(pluginAQueueingHintCalled, pod.Name)
+				return fwk.Queue, nil
+			},
+			PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+				return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod1", Namespace: "ns1"}}}, nil
+			},
+		},
+		{
+			PluginName: "pluginB",
+			QueueingHintFn: func(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (fwk.QueueingHint, error) {
+				pluginBQueueingHintCalled = append(pluginBQueueingHintCalled, pod.Name)
+				return fwk.Queue, nil
+			},
+			PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+				return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{{Name: "pod2", Namespace: "ns1"}}}, nil
+			},
+		},
+	}
+
+	q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m))
+
+	pod1 := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()
+	pod2 := st.MakePod().Name("pod2").Namespace("ns1").UID("2").Obj()
+
+	for _, pod := range []*v1.Pod{pod1, pod2} {
+		q.Add(ctx, pod)
+		entity, err := q.Pop(logger)
+		if err != nil {
+			t.Fatalf("Pop failed: %v", err)
+		}
+		pInfo := entity.(*framework.QueuedPodInfo)
+		pInfo.UnschedulablePlugins = sets.New[string]("pluginA", "pluginB")
+		if err := q.AddUnschedulablePodIfNotPresent(logger, pInfo, q.SchedulingCycle()); err != nil {
+			t.Fatalf("AddUnschedulablePodIfNotPresent failed: %v", err)
+		}
+	}
+
+	q.MoveAllToActiveOrBackoffQueue(logger, nodeAdd, nil, st.MakeNode().Name("node1").Obj(), nil)
+
+	// pluginA's QueueingHintFn should only be called for pod1.
+	if slices.Contains(pluginAQueueingHintCalled, "pod2") {
+		t.Errorf("pluginA QueueingHintFn should NOT be called for pod2, called for: %v", pluginAQueueingHintCalled)
+	}
+	if !slices.Contains(pluginAQueueingHintCalled, "pod1") {
+		t.Errorf("pluginA QueueingHintFn should be called for pod1, called for: %v", pluginAQueueingHintCalled)
+	}
+
+	// pluginB's QueueingHintFn should only be called for pod2.
+	if slices.Contains(pluginBQueueingHintCalled, "pod1") {
+		t.Errorf("pluginB QueueingHintFn should NOT be called for pod1, called for: %v", pluginBQueueingHintCalled)
+	}
+	if !slices.Contains(pluginBQueueingHintCalled, "pod2") {
+		t.Errorf("pluginB QueueingHintFn should be called for pod2, called for: %v", pluginBQueueingHintCalled)
+	}
+}
+
+/*
+Copyright 2017 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.
+*/

pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go

@@ -176,16 +176,24 @@ type nodeAllocation struct {
 
 // DynamicResources is a plugin that ensures that ResourceClaims are allocated.
 type DynamicResources struct {
-	enabled        bool
-	fts            feature.Features
-	filterTimeout  time.Duration
-	bindingTimeout time.Duration
-	fh             fwk.Handle
-	clientset      kubernetes.Interface
-	celCache       *cel.Cache
-	draManager     fwk.SharedDRAManager
+	enabled               bool
+	fts                   feature.Features
+	filterTimeout         time.Duration
+	bindingTimeout        time.Duration
+	fh                    fwk.Handle
+	clientset             kubernetes.Interface
+	celCache              *cel.Cache
+	draManager            fwk.SharedDRAManager
+	podIndexer            cache.Indexer
+	podResourceClaimIndex string
 }
 
+const (
+	// podResourceClaimIndexPrefix is the prefix for the index function which indexes pods by
+	// the ResourceClaims that they reference. The key for looking up pods is "<namespace>/<name>".
+	podResourceClaimIndexPrefix = "schedulerPodResourceClaim"
+)
+
 var (
 	ErrDeviceBindingTimeout = errors.New("device binding timeout")
 	ErrDeviceBindingFailed  = errors.New("device binding failed")
@@ -226,6 +234,15 @@ func New(ctx context.Context, plArgs runtime.Object, fh fwk.Handle, fts feature.
 		draManager: fh.SharedDRAManager(),
 	}
 
+	// Set up pod indexer for PreQueueingHint to look up pods by claim.
+	podInformer := fh.SharedInformerFactory().Core().V1().Pods()
+	podResourceClaimIndex := podResourceClaimIndexPrefix + "-" + fh.ProfileName()
+	if err := podInformer.Informer().AddIndexers(cache.Indexers{podResourceClaimIndex: podResourceClaimIndexFunc}); err != nil {
+		return nil, fmt.Errorf("failed to add pod resource claim indexer: %w", err)
+	}
+	pl.podIndexer = podInformer.Informer().GetIndexer()
+	pl.podResourceClaimIndex = podResourceClaimIndex
+
 	return pl, nil
 }
 
@@ -267,7 +284,7 @@ func (pl *DynamicResources) EventsToRegister(_ context.Context) ([]fwk.ClusterEv
 		// A new or updated node may make pods schedulable.
 		{Event: fwk.ClusterEvent{Resource: fwk.Node, ActionType: fwk.Add | fwk.UpdateNodeLabel | fwk.UpdateNodeAllocatable}},
 		// Allocation is tracked in ResourceClaims, so any changes may make the pods schedulable.
-		{Event: fwk.ClusterEvent{Resource: fwk.ResourceClaim, ActionType: fwk.Add | fwk.Update | fwk.Delete}, QueueingHintFn: pl.isSchedulableAfterClaimChange},
+		{Event: fwk.ClusterEvent{Resource: fwk.ResourceClaim, ActionType: fwk.Add | fwk.Update | fwk.Delete}, QueueingHintFn: pl.isSchedulableAfterClaimChange, PreQueueingHintFn: pl.preQueueingHint},
 		// Adding the ResourceClaim name to the pod status makes pods waiting for their ResourceClaim schedulable.
 		{Event: fwk.ClusterEvent{Resource: fwk.TargetPod, ActionType: fwk.UpdatePodGeneratedResourceClaim}, QueueingHintFn: pl.isSchedulableAfterTargetPodUpdate},
 		// A pod might be waiting for a class to get created or modified.
@@ -293,7 +310,68 @@ func (pl *DynamicResources) PreEnqueue(ctx context.Context, pod *v1.Pod) (status
 	return nil
 }
 
-// isSchedulableAfterClaimChange is invoked for add/update/delete claim events reported by
+// podResourceClaimIndexFunc is an index function that returns ResourceClaim keys
+// (namespace/name) for any ResourceClaim referenced by a pod.
+func podResourceClaimIndexFunc(obj interface{}) ([]string, error) {
+	pod, ok := obj.(*v1.Pod)
+	if !ok {
+		return nil, nil
+	}
+	keySet := sets.New[string]()
+	for _, podClaim := range pod.Spec.ResourceClaims {
+		claimName, _, err := resourceclaim.Name(pod, &podClaim)
+		if err != nil || claimName == nil {
+			continue
+		}
+		keySet.Insert(pod.Namespace + "/" + *claimName)
+	}
+	return keySet.UnsortedList(), nil
+}
+
+// preQueueingHint returns the pods affected by a ResourceClaim event.
+func (pl *DynamicResources) preQueueingHint(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+	if oldObj != nil && newObj != nil {
+		if oldClaim, ok := oldObj.(*resourceapi.ResourceClaim); ok {
+			if newClaim, ok := newObj.(*resourceapi.ResourceClaim); ok {
+				// Deallocation frees resources that may make other pods schedulable.
+				if oldClaim.Status.Allocation != nil && newClaim.Status.Allocation == nil {
+					return fwk.PreQueueingHintResult{AllPods: true}, nil
+				}
+			}
+		}
+	}
+
+	// Deletion of an allocated claim frees resources that may make other pods schedulable.
+	if newObj == nil && oldObj != nil {
+		if oldClaim, ok := oldObj.(*resourceapi.ResourceClaim); ok {
+			if oldClaim.Status.Allocation != nil {
+				return fwk.PreQueueingHintResult{AllPods: true}, nil
+			}
+		}
+	}
+
+	obj := newObj
+	if obj == nil {
+		obj = oldObj
+	}
+	claim, ok := obj.(*resourceapi.ResourceClaim)
+	if !ok {
+		return fwk.PreQueueingHintResult{AllPods: true}, nil
+	}
+	objs, err := pl.podIndexer.ByIndex(pl.podResourceClaimIndex, claim.Namespace+"/"+claim.Name)
+	if err != nil {
+		return fwk.PreQueueingHintResult{}, fmt.Errorf("failed to look up pods by claim %s/%s: %w", claim.Namespace, claim.Name, err)
+	}
+	pods := make([]types.NamespacedName, 0, len(objs))
+	for _, obj := range objs {
+		if pod, ok := obj.(*v1.Pod); ok {
+			pods = append(pods, types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace})
+		}
+	}
+	return fwk.PreQueueingHintResult{Pods: pods}, nil
+}
+
+// isSchedulableAfterClaimChange is invoked for add and update claim events reported by
 // an informer. It checks whether that change made a previously unschedulable
 // pod schedulable. It errs on the side of letting a pod scheduling attempt
 // happen.

pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go

@@ -5999,3 +5999,187 @@ func TestPodGroupPostFilter(t *testing.T) {
 		})
 	}
 }
+func TestPreQueueingHint(t *testing.T) {
+	tests := map[string]struct {
+		oldObj, newObj interface{}
+		wantKeys       sets.Set[string]
+	}{
+		"claim with no pod in indexer returns empty set": {
+			newObj: &resourceapi.ResourceClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "claim-1",
+					Namespace: "ns1",
+				},
+			},
+			wantKeys: sets.New[string](), // no pod in indexer
+		},
+		"shared claim returns nil": {
+			newObj: &resourceapi.ResourceClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "shared-claim",
+					Namespace: "ns1",
+				},
+			},
+			wantKeys: sets.New[string](), // no pod in indexer
+		},
+		"claim with no matching pods returns empty set": {
+			newObj: &resourceapi.ResourceClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "claim-2",
+					Namespace: "ns1",
+				},
+			},
+			wantKeys: sets.New[string](), // no pod in indexer
+		},
+		"delete event uses oldObj": {
+			oldObj: &resourceapi.ResourceClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "claim-3",
+					Namespace: "ns2",
+				},
+			},
+			newObj:   nil,
+			wantKeys: sets.New[string](), // no pod in indexer
+		},
+		"delete of allocated claim returns AllPods": {
+			oldObj: &resourceapi.ResourceClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "claim-allocated",
+					Namespace: "ns1",
+				},
+				Status: resourceapi.ResourceClaimStatus{
+					Allocation: &resourceapi.AllocationResult{},
+				},
+			},
+			newObj:   nil,
+			wantKeys: nil, // AllPods - deletion frees resources
+		},
+
+		"deallocation returns nil to evaluate all pods": {
+			oldObj: &resourceapi.ResourceClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "claim-4",
+					Namespace: "ns1",
+				},
+				Status: resourceapi.ResourceClaimStatus{
+					Allocation: &resourceapi.AllocationResult{},
+				},
+			},
+			newObj: &resourceapi.ResourceClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "claim-4",
+					Namespace: "ns1",
+				},
+			},
+			wantKeys: nil,
+		},
+		"non-claim object returns nil": {
+			newObj:   "not-a-claim",
+			wantKeys: nil,
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			logger := klog.Background()
+			pl := &DynamicResources{podIndexer: cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{podResourceClaimIndexPrefix + "-test": podResourceClaimIndexFunc}), podResourceClaimIndex: podResourceClaimIndexPrefix + "-test"}
+			got, err := pl.preQueueingHint(logger, tc.oldObj, tc.newObj)
+			if err != nil {
+				t.Fatalf("preQueueingHint returned unexpected error: %v", err)
+			}
+			if tc.wantKeys == nil {
+				// nil wantKeys means AllPods should be true
+				if !got.AllPods {
+					t.Errorf("expected AllPods=true, got %+v", got)
+				}
+			} else {
+				if got.AllPods {
+					t.Errorf("expected AllPods=false with specific pods, got AllPods=true")
+				}
+				gotPods := sets.New[string]()
+				for _, nn := range got.Pods {
+					gotPods.Insert(nn.Name + "_" + nn.Namespace)
+				}
+				wantPods := sets.New[string]()
+				for k := range tc.wantKeys {
+					wantPods.Insert(k)
+				}
+				if !gotPods.Equal(wantPods) {
+					t.Errorf("expected pods %v, got %v", wantPods, gotPods)
+				}
+			}
+		})
+	}
+}
+
+func TestPreQueueingHint_WithPodInIndexer(t *testing.T) {
+	// Verify that when a pod referencing a claim exists in the indexer,
+	// preQueueingHint returns that pod's key.
+	logger := klog.Background()
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{podResourceClaimIndexPrefix + "-test": podResourceClaimIndexFunc})
+
+	// Add a pod that references "claim-x" in namespace "ns1".
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-pod", Namespace: "ns1"},
+		Spec: v1.PodSpec{
+			ResourceClaims: []v1.PodResourceClaim{
+				{Name: "gpu", ResourceClaimName: new("claim-x")},
+			},
+		},
+	}
+	if err := indexer.Add(pod); err != nil {
+		t.Fatal(err)
+	}
+
+	pl := &DynamicResources{podIndexer: indexer, podResourceClaimIndex: podResourceClaimIndexPrefix + "-test"}
+
+	got, err := pl.preQueueingHint(logger, nil, &resourceapi.ResourceClaim{
+		ObjectMeta: metav1.ObjectMeta{Name: "claim-x", Namespace: "ns1"},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got.AllPods {
+		t.Fatal("expected AllPods=false, got true")
+	}
+	if len(got.Pods) != 1 || got.Pods[0].Name != "my-pod" || got.Pods[0].Namespace != "ns1" {
+		t.Errorf("expected [{my-pod ns1}], got %v", got.Pods)
+	}
+}
+
+func TestPreQueueingHint_SharedClaimMultiplePods(t *testing.T) {
+	// Verify that for a shared claim referenced by multiple pods,
+	// preQueueingHint returns all referencing pods.
+	logger := klog.Background()
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{podResourceClaimIndexPrefix + "-test": podResourceClaimIndexFunc})
+
+	// Add multiple pods referencing the same shared claim.
+	for _, name := range []string{"pod-a", "pod-b", "pod-c"} {
+		pod := &v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns1"},
+			Spec: v1.PodSpec{
+				ResourceClaims: []v1.PodResourceClaim{
+					{Name: "shared", ResourceClaimName: new("shared-claim")},
+				},
+			},
+		}
+		if err := indexer.Add(pod); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	pl := &DynamicResources{podIndexer: indexer, podResourceClaimIndex: podResourceClaimIndexPrefix + "-test"}
+
+	got, err := pl.preQueueingHint(logger, nil, &resourceapi.ResourceClaim{
+		ObjectMeta: metav1.ObjectMeta{Name: "shared-claim", Namespace: "ns1"},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got.AllPods {
+		t.Fatal("expected AllPods=false, got true")
+	}
+	if len(got.Pods) != 3 {
+		t.Errorf("expected 3 pods, got %d: %v", len(got.Pods), got.Pods)
+	}
+}

pkg/scheduler/framework/plugins/dynamicresources/prequeueing_race_test.go

@@ -0,0 +1,101 @@
+/*
+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 dynamicresources
+
+import (
+	"testing"
+
+	v1 "k8s.io/api/core/v1"
+	resourceapi "k8s.io/api/resource/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+)
+
+func TestPreQueueingHint_PodNotYetIndexed(t *testing.T) {
+	logger := klog.Background()
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{podResourceClaimIndexPrefix + "-test": podResourceClaimIndexFunc})
+
+	// Pod with template-based claim, status NOT yet updated
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-pod", Namespace: "ns1"},
+		Spec: v1.PodSpec{
+			ResourceClaims: []v1.PodResourceClaim{
+				{Name: "gpu", ResourceClaimTemplateName: new("gpu-template")},
+			},
+		},
+	}
+	if err := indexer.Add(pod); err != nil {
+		t.Fatal(err)
+	}
+
+	pl := &DynamicResources{podIndexer: indexer, podResourceClaimIndex: podResourceClaimIndexPrefix + "-test"}
+
+	// Claim allocation event - indexer cannot find pod by this claim name
+	got, err := pl.preQueueingHint(logger, nil, &resourceapi.ResourceClaim{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-pod-gpu-xyz", Namespace: "ns1"},
+		Status: resourceapi.ResourceClaimStatus{
+			Allocation: &resourceapi.AllocationResult{},
+		},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	// Known limitation: when pod's ResourceClaimStatuses isn't populated yet,
+	// the indexer cannot find the pod. The periodic flush rescues such pods.
+	if got.AllPods {
+		t.Logf("AllPods=true (safe fallback)")
+	} else if len(got.Pods) == 0 {
+		t.Logf("empty Pods (known race - pod rescued by flush)")
+	}
+}
+
+func TestPreQueueingHint_DeleteEventWithPodInIndexer(t *testing.T) {
+	// When an allocated claim is deleted and a pod referencing it exists
+	// in the indexer, AllPods is returned (deletion frees resources).
+	logger := klog.Background()
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{podResourceClaimIndexPrefix + "-test": podResourceClaimIndexFunc})
+
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-pod", Namespace: "ns1"},
+		Spec: v1.PodSpec{
+			ResourceClaims: []v1.PodResourceClaim{
+				{Name: "gpu", ResourceClaimName: new("deleted-claim")},
+			},
+		},
+	}
+	if err := indexer.Add(pod); err != nil {
+		t.Fatal(err)
+	}
+
+	pl := &DynamicResources{podIndexer: indexer, podResourceClaimIndex: podResourceClaimIndexPrefix + "-test"}
+
+	// Delete event for an allocated claim
+	got, err := pl.preQueueingHint(logger, &resourceapi.ResourceClaim{
+		ObjectMeta: metav1.ObjectMeta{Name: "deleted-claim", Namespace: "ns1"},
+		Status: resourceapi.ResourceClaimStatus{
+			Allocation: &resourceapi.AllocationResult{},
+		},
+	}, nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !got.AllPods {
+		t.Errorf("expected AllPods=true for delete of allocated claim, got %+v", got)
+	}
+}

pkg/scheduler/metrics/metric_recorder.go

@@ -151,6 +151,12 @@ type gaugeVecMetricKey struct {
 	labelValue string
 }
 
+type counterVecMetric struct {
+	metric      *metrics.CounterVec
+	labelValues []string
+	value       float64
+}
+
 // MetricAsyncRecorder records metric in a separate goroutine to avoid overhead in the critical path.
 type MetricAsyncRecorder struct {
 	// bufferCh is a channel that serves as a metrics buffer before the metricsRecorder goroutine reports it.
@@ -168,6 +174,9 @@ type MetricAsyncRecorder struct {
 	aggregatedInflightEventMetricLastFlushTime time.Time
 	aggregatedInflightEventMetricBufferCh      chan *gaugeVecMetric
 
+	// counterBufferCh is a channel for async counter increments.
+	counterBufferCh chan *counterVecMetric
+
 	// stopCh is used to stop the goroutine which periodically flushes metrics.
 	stopCh <-chan struct{}
 	// IsStoppedCh indicates whether the goroutine is stopped. It's used in tests only to make sure
@@ -178,6 +187,7 @@ type MetricAsyncRecorder struct {
 func NewMetricsAsyncRecorder(bufferSize int, interval time.Duration, stopCh <-chan struct{}) *MetricAsyncRecorder {
 	recorder := &MetricAsyncRecorder{
 		bufferCh:                      make(chan *histogramVecMetric, bufferSize),
+		counterBufferCh:               make(chan *counterVecMetric, bufferSize),
 		bufferSize:                    bufferSize,
 		interval:                      interval,
 		stopCh:                        stopCh,
@@ -208,6 +218,19 @@ func (r *MetricAsyncRecorder) ObserveQueueingHintDurationAsync(pluginName, event
 	r.observeMetricAsync(queueingHintExecutionDuration, value, pluginName, event, hint)
 }
 
+// ObserveCounterAsync increments a counter metric asynchronously.
+func (r *MetricAsyncRecorder) ObserveCounterAsync(m *metrics.CounterVec, value float64, labelValues ...string) {
+	newMetric := &counterVecMetric{
+		metric:      m,
+		labelValues: labelValues,
+		value:       value,
+	}
+	select {
+	case r.counterBufferCh <- newMetric:
+	default:
+	}
+}
+
 // ObserveInFlightEventsAsync observes the in_flight_events metric.
 //
 // Note that this function is not goroutine-safe;
@@ -267,6 +290,8 @@ func (r *MetricAsyncRecorder) FlushMetrics() {
 		select {
 		case m := <-r.bufferCh:
 			m.metric.WithLabelValues(m.labelValues...).Observe(m.value)
+		case m := <-r.counterBufferCh:
+			m.metric.WithLabelValues(m.labelValues...).Add(m.value)
 		default:
 			// no more value
 		}

pkg/scheduler/metrics/metrics.go

@@ -157,6 +157,7 @@ var (
 	PodSchedulingSLIDuration        *metrics.HistogramVec
 	PodSchedulingAttempts           *metrics.Histogram
 	PodScheduledAfterFlush          *metrics.Counter
+	PreQueueingHintEvaluations      *metrics.CounterVec
 	FrameworkExtensionPointDuration *metrics.HistogramVec
 	PluginExecutionDuration         *metrics.HistogramVec
 
@@ -381,6 +382,15 @@ func InitMetrics() {
 			StabilityLevel: metrics.ALPHA,
 		})
 
+	PreQueueingHintEvaluations = metrics.NewCounterVec(
+		&metrics.CounterOpts{
+			Subsystem:      SchedulerSubsystem,
+			Name:           "pre_queueing_hint_evaluations_total",
+			Help:           "Number of PreQueueingHint evaluations, labeled by plugin and result (all_pods or narrowed).",
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"plugin", "result"})
+
 	FrameworkExtensionPointDuration = metrics.NewHistogramVec(
 		&metrics.HistogramOpts{
 			Subsystem: SchedulerSubsystem,
@@ -687,6 +697,7 @@ func InitMetrics() {
 		StoreScheduleResultsDuration,
 		queueingHintExecutionDuration,
 		InFlightEvents,
+		PreQueueingHintEvaluations,
 	}
 }
 

pkg/scheduler/scheduler.go

@@ -514,8 +514,9 @@ func buildQueueingHintMap(ctx context.Context, es []fwk.EnqueueExtensions) (inte
 			}
 
 			queueingHintFn := &internalqueue.QueueingHintFunction{
-				PluginName:     e.Name(),
-				QueueingHintFn: fn,
+				PluginName:        e.Name(),
+				QueueingHintFn:    fn,
+				PreQueueingHintFn: event.PreQueueingHintFn,
 			}
 
 			if event.Event.Resource == fwk.Pod {

staging/src/k8s.io/kube-scheduler/framework/types.go

@@ -26,6 +26,7 @@ import (
 	schedulingv1beta1 "k8s.io/api/scheduling/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	ndf "k8s.io/component-helpers/nodedeclaredfeatures"
 	"k8s.io/klog/v2"
@@ -201,6 +202,9 @@ type ClusterEventWithHint struct {
 	// the scheduling of Pods will be always retried with backoff when this Event happens.
 	// (the same as Queue)
 	QueueingHintFn QueueingHintFn
+	// PreQueueingHintFn is called once per event to narrow the set of pods to evaluate with QueueingHint.
+	// If set, only pods identified in the returned PreQueueingHintResult are checked.
+	PreQueueingHintFn PreQueueingHintFn
 }
 
 // QueueingHintFn returns a hint that signals whether the event can make a Pod,
@@ -216,6 +220,24 @@ type ClusterEventWithHint struct {
 //   - `newObj` is nil if the event is delete event.
 type QueueingHintFn func(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (QueueingHint, error)
 
+// PreQueueingHintResult is the return value of PreQueueingHintFn.
+// When AllPods is true, all unschedulable pods are evaluated (existing behavior).
+// When AllPods is false, only pods listed in Pods are evaluated.
+type PreQueueingHintResult struct {
+	AllPods bool
+	Pods    []types.NamespacedName
+}
+
+// PreQueueingHintFn is called once per event before iterating unschedulable pods.
+// PreQueueingHintFn will not be called when the deleted object is unknown (both oldObj and newObj are nil).
+// It returns a PreQueueingHintResult indicating which pods should be evaluated
+// by QueueingHintFn for this event. If it returns an error, the error is logged
+// and the result is treated as AllPods (all unschedulable pods are evaluated).
+//
+// - oldObj is the old object in update events, or the deleted object in delete events.
+// - newObj is the new/added object. It is nil for delete events.
+type PreQueueingHintFn func(logger klog.Logger, oldObj, newObj interface{}) (PreQueueingHintResult, error)
+
 type QueueingHint int
 
 const (

test/compatibility_lifecycle/reference/feature_list.md

@@ -194,6 +194,7 @@
 | SchedulerAsyncAPICalls |  |  |  | 1.34– |  |  |  | [code](https://cs.k8s.io/?q=%5CbSchedulerAsyncAPICalls%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbSchedulerAsyncAPICalls%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | SchedulerAsyncPreemption | :ballot_box_with_check: 1.33+ |  | 1.32 | 1.33– |  |  |  | [code](https://cs.k8s.io/?q=%5CbSchedulerAsyncPreemption%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbSchedulerAsyncPreemption%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | SchedulerPopFromBackoffQ | :ballot_box_with_check: 1.33+ |  |  | 1.33– |  |  |  | [code](https://cs.k8s.io/?q=%5CbSchedulerPopFromBackoffQ%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbSchedulerPopFromBackoffQ%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
+| SchedulerPreQueueingHints | :ballot_box_with_check: 1.37+ |  |  | 1.37– |  |  |  | [code](https://cs.k8s.io/?q=%5CbSchedulerPreQueueingHints%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbSchedulerPreQueueingHints%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | SeparateCacheWatchRPC | :ballot_box_with_check: 1.28+ | :closed_lock_with_key: 1.36+ |  | 1.28–1.32 |  | 1.33– |  | [code](https://cs.k8s.io/?q=%5CbSeparateCacheWatchRPC%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbSeparateCacheWatchRPC%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | SeparateTaintEvictionController | :ballot_box_with_check: 1.29+ | :closed_lock_with_key: 1.34+ |  | 1.29–1.33 | 1.34– |  |  | [code](https://cs.k8s.io/?q=%5CbSeparateTaintEvictionController%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbSeparateTaintEvictionController%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | ServiceAccountNodeAudienceRestriction | :ballot_box_with_check: 1.33+ |  |  | 1.32– |  |  |  | [code](https://cs.k8s.io/?q=%5CbServiceAccountNodeAudienceRestriction%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbServiceAccountNodeAudienceRestriction%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |

test/compatibility_lifecycle/reference/versioned_feature_list.yaml

@@ -1811,6 +1811,12 @@
     lockToDefault: false
     preRelease: Beta
     version: "1.33"
+- name: SchedulerPreQueueingHints
+  versionedSpecs:
+  - default: true
+    lockToDefault: false
+    preRelease: Beta
+    version: "1.37"
 - name: SELinuxChangePolicy
   versionedSpecs:
   - default: false

test/integration/scheduler/prequeueinghint/main_test.go

@@ -0,0 +1,27 @@
+/*
+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 prequeueinghint
+
+import (
+	"testing"
+
+	"k8s.io/kubernetes/test/integration/framework"
+)
+
+func TestMain(m *testing.M) {
+	framework.EtcdMain(m.Run)
+}

test/integration/scheduler/prequeueinghint/prequeueinghint_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 prequeueinghint
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/klog/v2"
+	configv1 "k8s.io/kube-scheduler/config/v1"
+	fwk "k8s.io/kube-scheduler/framework"
+	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/scheduler"
+	configtesting "k8s.io/kubernetes/pkg/scheduler/apis/config/testing"
+	frameworkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
+	st "k8s.io/kubernetes/pkg/scheduler/testing"
+	testutils "k8s.io/kubernetes/test/integration/util"
+	"k8s.io/utils/ptr"
+)
+
+// preQueueingHintPlugin is a test plugin that implements PreQueueingHintFn
+// and tracks how many times its QueueingHintFn is called.
+type preQueueingHintPlugin struct {
+	queueingHintCalls atomic.Int64
+	// targetPod is the pod that PreQueueingHintFn will return.
+	targetPod types.NamespacedName
+}
+
+var _ fwk.FilterPlugin = &preQueueingHintPlugin{}
+var _ fwk.EnqueueExtensions = &preQueueingHintPlugin{}
+
+func (pl *preQueueingHintPlugin) Name() string {
+	return "PreQueueingHintTestPlugin"
+}
+
+func (pl *preQueueingHintPlugin) Filter(_ context.Context, _ fwk.CycleState, _ *v1.Pod, nodeInfo fwk.NodeInfo) *fwk.Status {
+	// Reject all pods - they need a node with a special label.
+	if nodeInfo.Node().Labels["enable-scheduling"] == "true" {
+		return nil
+	}
+	return fwk.NewStatus(fwk.Unschedulable, "node missing enable-scheduling label")
+}
+
+func (pl *preQueueingHintPlugin) EventsToRegister(_ context.Context) ([]fwk.ClusterEventWithHint, error) {
+	return []fwk.ClusterEventWithHint{
+		{
+			Event:          fwk.ClusterEvent{Resource: fwk.Node, ActionType: fwk.UpdateNodeLabel},
+			QueueingHintFn: pl.queueingHint,
+			PreQueueingHintFn: func(logger klog.Logger, oldObj, newObj interface{}) (fwk.PreQueueingHintResult, error) {
+				// Only return the target pod.
+				return fwk.PreQueueingHintResult{Pods: []types.NamespacedName{pl.targetPod}}, nil
+			},
+		},
+	}, nil
+}
+
+func (pl *preQueueingHintPlugin) queueingHint(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (fwk.QueueingHint, error) {
+	pl.queueingHintCalls.Add(1)
+	return fwk.Queue, nil
+}
+
+// TestPreQueueingHintNarrowsPodSet verifies that when PreQueueingHintFn is
+// registered and the feature gate is enabled, only the targeted pod's
+// QueueingHintFn is invoked on a matching event.
+func TestPreQueueingHintNarrowsPodSet(t *testing.T) {
+	tests := []struct {
+		name           string
+		featureEnabled bool
+		// With 3 unschedulable pods and PreQueueingHint targeting only pod-1,
+		// we expect QueueingHintFn to be called 1 time (enabled) or 3 times (disabled).
+		wantHintCalls int64
+	}{
+		{
+			name:           "feature enabled: only targeted pod evaluated",
+			featureEnabled: true,
+			wantHintCalls:  1,
+		},
+		{
+			name:           "feature disabled: all pods evaluated",
+			featureEnabled: false,
+			wantHintCalls:  3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerPreQueueingHints, tt.featureEnabled)
+
+			testCtx := testutils.InitTestAPIServer(t, "prequeueinghint", nil)
+
+			targetPod := types.NamespacedName{Name: "pod-1", Namespace: testCtx.NS.Name}
+			plugin := &preQueueingHintPlugin{targetPod: targetPod}
+
+			registry := frameworkruntime.Registry{
+				plugin.Name(): func(_ context.Context, _ runtime.Object, _ fwk.Handle) (fwk.Plugin, error) {
+					return plugin, nil
+				},
+			}
+
+			cfg := configtesting.V1ToInternalWithDefaults(t, configv1.KubeSchedulerConfiguration{
+				Profiles: []configv1.KubeSchedulerProfile{{
+					SchedulerName: ptr.To[string](v1.DefaultSchedulerName),
+					Plugins: &configv1.Plugins{
+						Filter: configv1.PluginSet{
+							Enabled: []configv1.Plugin{{Name: plugin.Name()}},
+						},
+					},
+				}},
+			})
+
+			testCtx = testutils.InitTestSchedulerWithOptions(t, testCtx, 0,
+				scheduler.WithProfiles(cfg.Profiles...),
+				scheduler.WithFrameworkOutOfTreeRegistry(registry),
+			)
+			testutils.SyncSchedulerInformerFactory(testCtx)
+			go testCtx.Scheduler.Run(testCtx.SchedulerCtx)
+
+			// Create a node without the required label (pods will be unschedulable).
+			node := st.MakeNode().Name("node-1").Label("kubernetes.io/hostname", "node-1").
+				Capacity(map[v1.ResourceName]string{v1.ResourceCPU: "2"}).Obj()
+			if _, err := testCtx.ClientSet.CoreV1().Nodes().Create(testCtx.Ctx, node, metav1.CreateOptions{}); err != nil {
+				t.Fatal(err)
+			}
+
+			// Create 3 pods that will all be unschedulable.
+			for i := 1; i <= 3; i++ {
+				pod := &v1.Pod{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      fmt.Sprintf("pod-%d", i),
+						Namespace: testCtx.NS.Name,
+					},
+					Spec: v1.PodSpec{
+						Containers: []v1.Container{{
+							Name:  "c",
+							Image: "pause",
+							Resources: v1.ResourceRequirements{
+								Requests: v1.ResourceList{v1.ResourceCPU: resource.MustParse("100m")},
+							},
+						}},
+					},
+				}
+				if _, err := testCtx.ClientSet.CoreV1().Pods(testCtx.NS.Name).Create(testCtx.Ctx, pod, metav1.CreateOptions{}); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			// Wait for all 3 pods to be attempted and become unschedulable.
+			if err := testutils.WaitForPodUnschedulableWithTimeout(testCtx.Ctx, testCtx.ClientSet, &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-3", Namespace: testCtx.NS.Name}}, 30*time.Second); err != nil {
+				t.Fatalf("pod-3 not unschedulable: %v", err)
+			}
+
+			// Reset the counter.
+			plugin.queueingHintCalls.Store(0)
+
+			// Trigger a node label update event (matches the plugin's EventsToRegister).
+			nodeCopy := node.DeepCopy()
+			nodeCopy.Labels["some-label"] = "some-value"
+			if _, err := testCtx.ClientSet.CoreV1().Nodes().Update(testCtx.Ctx, nodeCopy, metav1.UpdateOptions{}); err != nil {
+				t.Fatal(err)
+			}
+
+			// Wait for the scheduler to process the event.
+			err := wait.PollUntilContextTimeout(testCtx.Ctx, 100*time.Millisecond, 10*time.Second, true, func(ctx context.Context) (bool, error) {
+				return plugin.queueingHintCalls.Load() == tt.wantHintCalls, nil
+			})
+			if err != nil {
+				t.Errorf("timed out waiting for QueueingHintFn calls: got %d, want %d", plugin.queueingHintCalls.Load(), tt.wantHintCalls)
+			}
+		})
+	}
+}