Repository: kubernetes/kubernetes (language: go)

Issue:
[Bug] Scheduler fails to schedule a pod due to a race condition
### What happened?

A bug in the scheduler increases the time spent on scheduling a pod **from <1 second to 5 minutes**.

We discovered the bug when repeating the steps described in a fixed bug report [#106780](https://github.com/kubernetes/kubernetes/issues/106780). The setting and steps are as follows:

We have 1 node with 32Gi memory, and 3 pods to schedule: 
p1 / request 10Gi memory / low-priority,
p2 / request 25Gi memory / medium-priority,
p3 / request 20Gi memory / high-priority.

We perform the following steps:
1. add `p1` and wait until `p1` is running
2. add `p2` and wait until `p1` is terminating; this is because there are not enough resources to host both pods and `p2` has a higher priority.
**Note that at this point, `p2` is not running yet.**
3. add `p3` and wait until `p3` is running; `p2` is pending because `p2` and `p3` cannot coexist on `node0`.
4. re-add `p1` and wait until `p1` is running; `p1` should be able to run since there are enough resources to host both pods: 10 + 20 < 32

Interestingly, we find that if step 4 happened immediately after step 3, then `p1` fails to get scheduled with the reason `Insufficient Memory` and eventually gets scheduled after 5 minutes. If there is some short period between step 3 and step 4 (say 2 seconds), then `p1` is scheduled properly immediately.

#### What is the root cause?

After code and log inspections (see below), we have found the root cause of this bug: it's caused by the race condition between (A) `p1` is re-added and handled by the scheduler and (B) `p2`'s `nominated_node_name` is cleared. If A happens before B, then the bug occurs.

#### Why does 2 second make the difference?

The reason is that `p2`'s `nominated_node_name` is cleared *when it (a nominated pod) is scheduled again and failed*. But as the `p2`'s first schedule failed (need to wait for preemption to finish), it's put to BackoffQueue and need to wait for a few seconds before it's put back to ActionQueue and scheduled again.

So 
1. If `p1` is added before `p2` is scheduled again (without 2s sleep), it will **fail** as `p2`'s `nomianted_node_name` is not cleared.
2. But if we add the 2s sleep, `p2` is scheduled again and its `nominated_node_name` is cleared before `p1` is added. In this case, `p1` will be **schedulable** to node0.

<details><summary>Some scheduler logs of the bug-free and buggy trace</summary>
The lines starting with # are logged just before scheduling p1 and p2. You can notice that their order is reversed in the two traces.

Bug-free trace (B happens before A):
```markdown
# with sleep(2s)
I0814 14:19:47.939731   32287 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p3"
I0814 14:19:47.939756   32287 schedule_one.go:83] "About to try and schedule pod" pod="default/p3"
I0814 14:19:47.939768   32287 schedule_one.go:96] "Attempting to schedule pod" pod="default/p3"
I0814 14:19:47.939878   32287 default_binder.go:53] "Attempting to bind pod to node" logger="Bind.DefaultBinder" pod="default/p3" node="node0"
I0814 14:19:47.944902   32287 eventhandlers.go:313] "Delete event for scheduled pod" pod="default/p1"
I0814 14:19:47.948208   32287 scheduling_queue.go:1312] "Pod moved to an internal scheduling queue" pod="default/p2" event="AssignedPodDelete" queue="Backoff" hint=1
I0814 14:19:47.948799   32287 schedule_one.go:314] "Successfully bound pod to node" pod="default/p3" node="node0" evaluatedNodes=1 feasibleNodes=1
I0814 14:19:47.948891   32287 eventhandlers.go:201] "Delete event for unscheduled pod" pod="default/p3"
I0814 14:19:47.948901   32287 eventhandlers.go:231] "Add event for scheduled pod" pod="default/p3"
I0814 14:19:47.956662   32287 eventhandlers.go:268] "Update event for scheduled pod" pod="default/p3"
I0814 14:19:49.047040   32287 schedule_one.go:83] "About to try and schedule pod" pod="default/p2"
# I0814 14:19:49.047238   32287 schedule_one.go:96] "Attempting to schedule pod" pod="default/p2"
I0814 14:19:49.047886   32287 schedule_one.go:1055] "Unable to schedule pod; no fit; waiting" pod="default/p2" err="0/1 nodes are available: 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod."
I0814 14:19:49.048895   32287 schedule_one.go:1122] "Updating pod condition" pod="default/p2" conditionType="PodScheduled" conditionStatus="False" conditionReason="Unschedulable"
I0814 14:19:49.064048   32287 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p2"
I0814 14:19:49.978428   32287 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p1"
I0814 14:19:49.978562   32287 schedule_one.go:83] "About to try and schedule pod" pod="default/p1"
# I0814 14:19:49.978580   32287 schedule_one.go:96] "Attempting to schedule pod" pod="default/p1"
I0814 14:19:49.978879   32287 default_binder.go:53] "Attempting to bind pod to node" logger="Bind.DefaultBinder" pod="default/p1" node="node0"
I0814 14:19:49.985438   32287 eventhandlers.go:201] "Delete event for unscheduled pod" pod="default/p1"
I0814 14:19:49.985476   32287 eventhandlers.go:231] "Add event for scheduled pod" pod="default/p1"
I0814 14:19:49.985747   32287 schedule_one.go:314] "Successfully bound pod to node" pod="default/p1" node="node0" evaluatedNodes=1 feasibleNodes=1
```

Buggy trace (A happens before B):
```markdown
# without sleep(2s)
I0814 14:17:47.151445   31615 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p3"
I0814 14:17:47.151515   31615 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p2"
I0814 14:17:47.151802   31615 schedule_one.go:83] "About to try and schedule pod" pod="default/p3"
I0814 14:17:47.151812   31615 schedule_one.go:96] "Attempting to schedule pod" pod="default/p3"
I0814 14:17:47.151926   31615 default_binder.go:53] "Attempting to bind pod to node" logger="Bind.DefaultBinder" pod="default/p3" node="node0"
I0814 14:17:47.156395   31615 eventhandlers.go:313] "Delete event for scheduled pod" pod="default/p1"
I0814 14:17:47.156444   31615 scheduling_queue.go:1312] "Pod moved to an internal scheduling queue" pod="default/p2" event="AssignedPodDelete" queue="Backoff" hint=1
I0814 14:17:47.160464   31615 eventhandlers.go:201] "Delete event for unscheduled pod" pod="default/p3"
I0814 14:17:47.160479   31615 eventhandlers.go:231] "Add event for scheduled pod" pod="default/p3"
I0814 14:17:47.160514   31615 schedule_one.go:314] "Successfully bound pod to node" pod="default/p3" node="node0" evaluatedNodes=1 feasibleNodes=1
I0814 14:17:47.168539   31615 eventhandlers.go:268] "Update event for scheduled pod" pod="default/p3"
I0814 14:17:47.177608   31615 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p1"
I0814 14:17:47.177642   31615 schedule_one.go:83] "About to try and schedule pod" pod="default/p1"
# I0814 14:17:47.177650   31615 schedule_one.go:96] "Attempting to schedule pod" pod="default/p1"
I0814 14:17:47.177775   31615 schedule_one.go:1055] "Unable to schedule pod; no fit; waiting" pod="default/p1" err="0/1 nodes are available: 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod."
I0814 14:17:47.177802   31615 schedule_one.go:1122] "Updating pod condition" pod="default/p1" conditionType="PodScheduled" conditionStatus="False" conditionReason="Unschedulable"
I0814 14:17:47.187472   31615 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p1"
I0814 14:17:48.138665   31615 schedule_one.go:83] "About to try and schedule pod" pod="default/p2"
# I0814 14:17:48.139950   31615 schedule_one.go:96] "Attempting to schedule pod" pod="default/p2"
I0814 14:17:48.140305   31615 schedule_one.go:1055] "Unable to schedule pod; no fit; waiting" pod="default/p2" err="0/1 nodes are available: 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod."
I0814 14:17:48.140582   31615 schedule_one.go:1122] "Updating pod condition" pod="default/p2" conditionType="PodScheduled" conditionStatus="False" conditionReason="Unschedulable"
I0814 14:17:48.156537   31615 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p2"
```
</details>

#### How to fix it?

1. A potential fix is:
Register all plugins that need to be aware of pods to events triggered by updates to a pod's `nominated_node_name`. This way, a failed pod will be retried as soon as any pod's `nominated_node_name` is cleared.

2. If this is too radical and can cause performance issues, another solution is: 
When a pod is bound to a node with nominated pods, either 
a) re-check the previously nominated pods (that nominated to this node), 
b) or clear the `nominated_node_name` of these nominated pods.
In this way, upcoming pods won't fail unreasonably and won't have to wait 5 minutes for the next retry.


### What did you expect to happen?

`p1` should be immediately scheduled.


### How can we reproduce it (as minimally and precisely as possible)?

We're using [kwok](https://kwok.sigs.k8s.io/) to reproduce this issue.

```yaml
# node0.yaml
apiVersion: v1
kind: Node
metadata:
  name: node0
  labels:
    kubernetes.io/hostname: node0
status:
  allocatable:
    cpu: "32"
    memory: "32Gi"
    pods: "110"
  capacity:
    cpu: "32"
    memory: "32Gi"
    pods: "110"
```

```yaml
# p1.yaml
apiVersion: v1
kind: Pod
metadata:
  name: p1
spec:
  containers:
  - name: p1-container
    image: nginx
    resources:
      requests:
        memory: "10Gi"

  priorityClassName: low-priority
```

```yaml
# p2.yaml
apiVersion: v1
kind: Pod
metadata:
  name: p2
spec:
  containers:
  - name: p2-container
    image: nginx
    resources:
      requests:
        memory: "25Gi"
  priorityClassName: medium-priority
```

```yaml
# p3.yaml
apiVersion: v1
kind: Pod
metadata:
  name: p3
spec:
  containers:
  - name: p3-container
    image: nginx
    resources:
      requests:
        memory: "20Gi"
  priorityClassName: high-priority
```

```yaml
# priority_class.yaml
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: high-priority
value: 1000000
globalDefault: false
description: "This priority class should be used for XYZ service pods only."
---
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: medium-priority
value: 500000
globalDefault: false
description: "This priority class should be used for XYZ service pods only."
---
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: low-priority
value: 0
globalDefault: false
description: "This priority class should be used for XYZ service pods only."
```


```python
# To run this reproduction script, please create a new KWOK cluster first.
# kwokctl create cluster --v DEBUG
# Also need to install the kubernetes python client library,
# and place all yaml files in the same folder with the script.
# ---
# To check the scheduler logs:
# kwokctl logs kube-scheduler
# ---
# To reproduce the bug case, please comment out `sleep(2)` between step 5 and step 6.
# To reproduce the normal case, please add `sleep(2)` between step 5 and step 6.
import unittest
import time
import shutil
import logging
from time import sleep
from os import path, makedirs
from logging import getLogger
from kubernetes import config, watch
from kubernetes.client import *
from kubernetes.utils import *
from time import strftime

logger = getLogger(__name__)
config.load_kube_config()
v1 = CoreV1Api()
k8s_cli = ApiClient()


log_dir = path.dirname(__file__)

if __name__ == "__main__":
    log_path = path.join(log_dir, f"issue-106780-reproduction-{strftime('%Y-%m-%d-%H-%M-%S')}")
    shutil.rmtree(log_path, ignore_errors=True)
    makedirs(log_path, exist_ok=True)
    log_file = path.join(log_path, 'reproduce.log')
    logging.basicConfig(level=logging.INFO,
                        filemode='w',
                        filename=log_file,
                        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                        datefmt='%H:%M:%S')
    node0_yaml = path.join(path.dirname(__file__), 'node0.yaml')
    p1_yaml = path.join(path.dirname(__file__), 'p1.yaml')
    p2_yaml = path.join(path.dirname(__file__), 'p2.yaml')
    p3_yaml = path.join(path.dirname(__file__), 'p3.yaml')
    priority_class_yaml = path.join(path.dirname(__file__), 'priority_class.yaml')
    w = watch.Watch()

    # 1. create priority class
    create_from_yaml(k8s_client=k8s_cli, yaml_file=priority_class_yaml)
    logger.info(f"PriorityClass created")

    # 2. add node0
    create_from_yaml(k8s_client=k8s_cli, yaml_file=node0_yaml)
    logger.info(f"Node: node0 created")

    # 3. add p1
    create_from_yaml(k8s_client=k8s_cli, yaml_file=p1_yaml)
    logger.info(f"Pod: p1 added")

    # 3. check p1 is running
    p1_running = False
    for event in w.stream(v1.list_namespaced_pod,
                          field_selector=f'metadata.name=p1',
                          namespace="default",
                          timeout_seconds=int(5)):
        p1 = event['object']
        if event['object'].status.phase == "Running":
            p1_running = True
            logger.info(f"Pod: {p1.metadata.name} scheduled on node {p1.spec.node_name}")
            break
    assert p1_running

    # 4. add p2
    create_from_yaml(k8s_client=k8s_cli, yaml_file=p2_yaml)
    logger.info(f"Pod: p2 added")

    # 4. check p1 is terminating (by p2)
    for event in w.stream(v1.list_namespaced_pod,
                          field_selector=f'metadata.name=p1',
                          namespace="default",
                          timeout_seconds=int(3)):
        logger.info(f"Pod: {event['object'].metadata.name}, event: {event['type']}")
        logger.info(
            f"deletion_timestamp: {event['object'].metadata.deletion_timestamp}, status: {event['object'].status.phase}")
        if event['object'].metadata.deletion_timestamp is not None and event['object'].status.phase in (
        'Pending', 'Running'):
            logger.info(f"Pod: {p1.metadata.name} is terminating")
            break

    # 5. add p3 and wait for p3 running (p3 will terminate p2 / let p2 pending)
    create_from_yaml(k8s_client=k8s_cli, yaml_file=p3_yaml)
    p3_running = False
    for event in w.stream(v1.list_namespaced_pod,
                          field_selector=f'metadata.name=p3',
                          namespace="default",
                          timeout_seconds=int(5)):
        p3 = event['object']
        if p3.status.phase == "Running":
            logger.info(f"Pod: {p3.metadata.name} scheduled on node p3")
            p3_running = True
            break
    assert p3_running

    # sleep(2)  # -> without this on 1.30.2 will cause pod scheduled failed at the first try

    # 6. reapply and check p1 schedulable
    create_from_yaml(k8s_client=k8s_cli, yaml_file=p1_yaml)
    tic = time.time()
    logger.info(f"Pod: p1 re-applied")

    # 7. watch pod1 events and wait it to be scheduled
    p1_running = False
    for event in w.stream(v1.list_namespaced_pod,
                          field_selector=f'metadata.name=p1',
                          namespace="default",
                          timeout_seconds=int(600)):
        p1 = event['object']
        if p1.status.phase == "Running":
            p1_running = True
            logger.info(f"Pod: {p1.metadata.name} scheduled on node {p1.spec.node_name}")
            break

    if not p1_running:
        logger.error(f"Pod: {p1.metadata.name} is not scheduled in 600s")
        assert False
    else:
        logger.info(f"Pod: {p1.metadata.name}, " + f"time between add and scheduled: {time.time() - tic}")


```


### Anything else we need to know?

Tested on 1.30.2

/sig bugs scheduling

### Kubernetes version

<details>

Tested on 1.30.2

</details>


### Cloud provider

<details>

</details>


### OS version

<details>

```console
# On Linux:
$ cat /etc/os-release
# paste output here
$ uname -a
# paste output here

# On Windows:
C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
# paste output here
```

</details>


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>



Discussion / hints:
@scheduler-tester: The label(s) `sig/bugs` cannot be applied, because the repository doesn't have them.

<details>

In response to [this](https://github.com/kubernetes/kubernetes/issues/126700):

>### What happened?
>
>A bug in the scheduler increases the time spent on scheduling a pod **from <1 second to 5 minutes**.
>
>We discovered the bug when repeating the steps described in a fixed bug report [#106780](https://github.com/kubernetes/kubernetes/issues/106780). The setting and steps are as follows:
>
>We have 1 node with 32Gi memory, and 3 pods to schedule: 
>p1 / request 10Gi memory / low-priority,
>p2 / request 25Gi memory / medium-priority,
>p3 / request 20Gi memory / high-priority.
>
>We perform the following steps:
>1. add `p1` and wait until `p1` is running
>2. add `p2` and wait until `p1` is terminating; this is because there are not enough resources to host both pods and `p2` has a higher priority.
>**Note that at this point, `p2` is not running yet.**
>3. add `p3` and wait until `p3` is running; `p2` is pending because `p2` and `p3` cannot coexist on `node0`.
>4. re-add `p1` and wait until `p1` is running; `p1` should be able to run since there are enough resources to host both pods: 10 + 20 < 32
>
>Interestingly, we find that if step 4 happened immediately after step 3, then `p1` fails to get scheduled with the reason `Insufficient Memory` and eventually gets scheduled after 5 minutes. If there is some short period between step 3 and step 4 (say 2 seconds), then `p1` is scheduled properly immediately.
>
>#### What is the root cause?
>
>After code and log inspections (see below), we have found the root cause of this bug: it's caused by the race condition between (A) `p1` is re-added and handled by the scheduler and (B) `p2`'s `nominated_node_name` is cleared. If A happens before B, then the bug occurs.
>
>#### Why does 2 second make the difference?
>
>The reason is that `p2`'s `nominated_node_name` is cleared *when it (a nominated pod) is scheduled again and failed*. But as the `p2`'s first schedule failed (need to wait for preemption to finish), it's put to BackoffQueue and need to wait for a few seconds before it's put back to ActionQueue and scheduled again.
>
>So 
>1. If `p1` is added before `p2` is scheduled again (without 2s sleep), it will **fail** as `p2`'s `nomianted_node_name` is not cleared.
>2. But if we add the 2s sleep, `p2` is scheduled again and its `nominated_node_name` is cleared before `p1` is added. In this case, `p1` will be **schedulable** to node0.
>
><details><summary>Some scheduler logs of the bug-free and buggy trace</summary>
>The lines starting with # are logged just before scheduling p1 and p2. You can notice that their order is reversed in the two traces.
>
>Bug-free trace (B happens before A):
>```markdown
># with sleep(2s)
>I0814 14:19:47.939731   32287 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p3"
>I0814 14:19:47.939756   32287 schedule_one.go:83] "About to try and schedule pod" pod="default/p3"
>I0814 14:19:47.939768   32287 schedule_one.go:96] "Attempting to schedule pod" pod="default/p3"
>I0814 14:19:47.939878   32287 default_binder.go:53] "Attempting to bind pod to node" logger="Bind.DefaultBinder" pod="default/p3" node="node0"
>I0814 14:19:47.944902   32287 eventhandlers.go:313] "Delete event for scheduled pod" pod="default/p1"
>I0814 14:19:47.948208   32287 scheduling_queue.go:1312] "Pod moved to an internal scheduling queue" pod="default/p2" event="AssignedPodDelete" queue="Backoff" hint=1
>I0814 14:19:47.948799   32287 schedule_one.go:314] "Successfully bound pod to node" pod="default/p3" node="node0" evaluatedNodes=1 feasibleNodes=1
>I0814 14:19:47.948891   32287 eventhandlers.go:201] "Delete event for unscheduled pod" pod="default/p3"
>I0814 14:19:47.948901   32287 eventhandlers.go:231] "Add event for scheduled pod" pod="default/p3"
>I0814 14:19:47.956662   32287 eventhandlers.go:268] "Update event for scheduled pod" pod="default/p3"
>I0814 14:19:49.047040   32287 schedule_one.go:83] "About to try and schedule pod" pod="default/p2"
># I0814 14:19:49.047238   32287 schedule_one.go:96] "Attempting to schedule pod" pod="default/p2"
>I0814 14:19:49.047886   32287 schedule_one.go:1055] "Unable to schedule pod; no fit; waiting" pod="default/p2" err="0/1 nodes are available: 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod."
>I0814 14:19:49.048895   32287 schedule_one.go:1122] "Updating pod condition" pod="default/p2" conditionType="PodScheduled" conditionStatus="False" conditionReason="Unschedulable"
>I0814 14:19:49.064048   32287 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p2"
>I0814 14:19:49.978428   32287 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p1"
>I0814 14:19:49.978562   32287 schedule_one.go:83] "About to try and schedule pod" pod="default/p1"
># I0814 14:19:49.978580   32287 schedule_one.go:96] "Attempting to schedule pod" pod="default/p1"
>I0814 14:19:49.978879   32287 default_binder.go:53] "Attempting to bind pod to node" logger="Bind.DefaultBinder" pod="default/p1" node="node0"
>I0814 14:19:49.985438   32287 eventhandlers.go:201] "Delete event for unscheduled pod" pod="default/p1"
>I0814 14:19:49.985476   32287 eventhandlers.go:231] "Add event for scheduled pod" pod="default/p1"
>I0814 14:19:49.985747   32287 schedule_one.go:314] "Successfully bound pod to node" pod="default/p1" node="node0" evaluatedNodes=1 feasibleNodes=1
>```
>
>Buggy trace (A happens before B):
>```markdown
># without sleep(2s)
>I0814 14:17:47.151445   31615 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p3"
>I0814 14:17:47.151515   31615 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p2"
>I0814 14:17:47.151802   31615 schedule_one.go:83] "About to try and schedule pod" pod="default/p3"
>I0814 14:17:47.151812   31615 schedule_one.go:96] "Attempting to schedule pod" pod="default/p3"
>I0814 14:17:47.151926   31615 default_binder.go:53] "Attempting to bind pod to node" logger="Bind.DefaultBinder" pod="default/p3" node="node0"
>I0814 14:17:47.156395   31615 eventhandlers.go:313] "Delete event for scheduled pod" pod="default/p1"
>I0814 14:17:47.156444   31615 scheduling_queue.go:1312] "Pod moved to an internal scheduling queue" pod="default/p2" event="AssignedPodDelete" queue="Backoff" hint=1
>I0814 14:17:47.160464   31615 eventhandlers.go:201] "Delete event for unscheduled pod" pod="default/p3"
>I0814 14:17:47.160479   31615 eventhandlers.go:231] "Add event for scheduled pod" pod="default/p3"
>I0814 14:17:47.160514   31615 schedule_one.go:314] "Successfully bound pod to node" pod="default/p3" node="node0" evaluatedNodes=1 feasibleNodes=1
>I0814 14:17:47.168539   31615 eventhandlers.go:268] "Update event for scheduled pod" pod="default/p3"
>I0814 14:17:47.177608   31615 eventhandlers.go:149] "Add event for unscheduled pod" pod="default/p1"
>I0814 14:17:47.177642   31615 schedule_one.go:83] "About to try and schedule pod" pod="default/p1"
># I0814 14:17:47.177650   31615 schedule_one.go:96] "Attempting to schedule pod" pod="default/p1"
>I0814 14:17:47.177775   31615 schedule_one.go:1055] "Unable to schedule pod; no fit; waiting" pod="default/p1" err="0/1 nodes are available: 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod."
>I0814 14:17:47.177802   31615 schedule_one.go:1122] "Updating pod condition" pod="default/p1" conditionType="PodScheduled" conditionStatus="False" conditionReason="Unschedulable"
>I0814 14:17:47.187472   31615 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p1"
>I0814 14:17:48.138665   31615 schedule_one.go:83] "About to try and schedule pod" pod="default/p2"
># I0814 14:17:48.139950   31615 schedule_one.go:96] "Attempting to schedule pod" pod="default/p2"
>I0814 14:17:48.140305   31615 schedule_one.go:1055] "Unable to schedule pod; no fit; waiting" pod="default/p2" err="0/1 nodes are available: 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod."
>I0814 14:17:48.140582   31615 schedule_one.go:1122] "Updating pod condition" pod="default/p2" conditionType="PodScheduled" conditionStatus="False" conditionReason="Unschedulable"
>I0814 14:17:48.156537   31615 eventhandlers.go:174] "Update event for unscheduled pod" pod="default/p2"
>```
></details>
>
>#### How to fix it?
>
>A potential fix is to register all plugins that need to be aware of pods to events triggered by updates to a pod's `nominated_node_name`. This way, a failed pod will be retried as soon as any pod's `nominated_node_name` is cleared.
>
>If this is too radical and can cause performance issues, another solution is: When a pod is bound to a node with nominated pods, either re-check the previously nominated pods (that nominated to this node) or clear the `nominated_node_name` of these nominated pods. In this way, upcoming pods won't fail unreasonably and won't have to wait 5 minutes for the next retry.
>
>
>### What did you expect to happen?
>
>`p1` should be immediately scheduled.
>
>
>### How can we reproduce it (as minimally and precisely as possible)?
>
>We're using [kwok](https://kwok.sigs.k8s.io/) to reproduce this issue.
>
>```yaml
># node0.yaml
>apiVersion: v1
>kind: Node
>metadata:
>  name: node0
>  labels:
>    kubernetes.io/hostname: node0
>status:
>  allocatable:
>    cpu: "32"
>    memory: "32Gi"
>    pods: "110"
>  capacity:
>    cpu: "32"
>    memory: "32Gi"
>    pods: "110"
>```
>
>```yaml
># p1.yaml
>apiVersion: v1
>kind: Pod
>metadata:
>  name: p1
>spec:
>  containers:
>  - name: p1-container
>    image: nginx
>    resources:
>      requests:
>        memory: "10Gi"
>
>  priorityClassName: low-priority
>```
>
>```yaml
># p2.yaml
>apiVersion: v1
>kind: Pod
>metadata:
>  name: p2
>spec:
>  containers:
>  - name: p2-container
>    image: nginx
>    resources:
>      requests:
>        memory: "25Gi"
>  priorityClassName: medium-priority
>```
>
>```yaml
># p3.yaml
>apiVersion: v1
>kind: Pod
>metadata:
>  name: p3
>spec:
>  containers:
>  - name: p3-container
>    image: nginx
>    resources:
>      requests:
>        memory: "20Gi"
>  priorityClassName: high-priority
>```
>
>```yaml
># priority_class.yaml
>apiVersion: scheduling.k8s.io/v1
>kind: PriorityClass
>metadata:
>  name: high-priority
>value: 1000000
>globalDefault: false
>description: "This priority class should be used for XYZ service pods only."
>---
>apiVersion: scheduling.k8s.io/v1
>kind: PriorityClass
>metadata:
>  name: medium-priority
>value: 500000
>globalDefault: false
>description: "This priority class should be used for XYZ service pods only."
>---
>apiVersion: scheduling.k8s.io/v1
>kind: PriorityClass
>metadata:
>  name: low-priority
>value: 0
>globalDefault: false
>description: "This priority class should be used for XYZ service pods only."
>```
>
>
>```python
># To run this reproduction script, please create a new KWOK cluster first.
># kwokctl create cluster --v DEBUG
># Also need to install the kubernetes python client library,
># and place all yaml files in the same folder with the script.
># ---
># To check the scheduler logs:
># kwokctl logs kube-scheduler
>import unittest
>import time
>import shutil
>import logging
>from time import sleep
>from os import path, makedirs
>from logging import getLogger
>from kubernetes import config, watch
>from kubernetes.client import *
>from kubernetes.utils import *
>from time import strftime
>
>logger = getLogger(__name__)
>config.load_kube_config()
>v1 = CoreV1Api()
>k8s_cli = ApiClient()
>
>
>log_dir = path.dirname(__file__)
>
>if __name__ == "__main__":
>    log_path = path.join(log_dir, f"issue-106780-reproduction-{strftime('%Y-%m-%d-%H-%M-%S')}")
>    shutil.rmtree(log_path, ignore_errors=True)
>    makedirs(log_path, exist_ok=True)
>    log_file = path.join(log_path, 'reproduce.log')
>    logging.basicConfig(level=logging.INFO,
>                        filemode='w',
>                        filename=log_file,
>                        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
>                        datefmt='%H:%M:%S')
>    node0_yaml = path.join(path.dirname(__file__), 'node0.yaml')
>    p1_yaml = path.join(path.dirname(__file__), 'p1.yaml')
>    p2_yaml = path.join(path.dirname(__file__), 'p2.yaml')
>    p3_yaml = path.join(path.dirname(__file__), 'p3.yaml')
>    priority_class_yaml = path.join(path.dirname(__file__), 'priority_class.yaml')
>    w = watch.Watch()
>
>    # 1. create priority class
>    create_from_yaml(k8s_client=k8s_cli, yaml_file=priority_class_yaml)
>    logger.info(f"PriorityClass created")
>
>    # 2. add node0
>    create_from_yaml(k8s_client=k8s_cli, yaml_file=node0_yaml)
>    logger.info(f"Node: node0 created")
>
>    # 3. add p1
>    create_from_yaml(k8s_client=k8s_cli, yaml_file=p1_yaml)
>    logger.info(f"Pod: p1 added")
>
>    # 3. check p1 is running
>    p1_running = False
>    for event in w.stream(v1.list_namespaced_pod,
>                          field_selector=f'metadata.name=p1',
>                          namespace="default",
>                          timeout_seconds=int(5)):
>        p1 = event['object']
>        if event['object'].status.phase == "Running":
>            p1_running = True
>            logger.info(f"Pod: {p1.metadata.name} scheduled on node {p1.spec.node_name}")
>            break
>    assert p1_running
>
>    # 4. add p2
>    create_from_yaml(k8s_client=k8s_cli, yaml_file=p2_yaml)
>    logger.info(f"Pod: p2 added")
>
>    # 4. check p1 is terminating (by p2)
>    for event in w.stream(v1.list_namespaced_pod,
>                          field_selector=f'metadata.name=p1',
>                          namespace="default",
>                          timeout_seconds=int(3)):
>        logger.info(f"Pod: {event['object'].metadata.name}, event: {event['type']}")
>        logger.info(
>            f"deletion_timestamp: {event['object'].metadata.deletion_timestamp}, status: {event['object'].status.phase}")
>        if event['object'].metadata.deletion_timestamp is not None and event['object'].status.phase in (
>        'Pending', 'Running'):
>            logger.info(f"Pod: {p1.metadata.name} is terminating")
>            break
>
>    # 5. add p3 and wait for p3 running (p3 will terminate p2 / let p2 pending)
>    create_from_yaml(k8s_client=k8s_cli, yaml_file=p3_yaml)
>    p3_running = False
>    for event in w.stream(v1.list_namespaced_pod,
>                          field_selector=f'metadata.name=p3',
>                          namespace="default",
>                          timeout_seconds=int(5)):
>        p3 = event['object']
>        if p3.status.phase == "Running":
>            logger.info(f"Pod: {p3.metadata.name} scheduled on node p3")
>            p3_running = True
>            break
>    assert p3_running
>
>    sleep(2)  # -> without this on 1.30.2 will cause pod scheduled failed at the first try
>
>    # 6. reapply and check p1 schedulable
>    create_from_yaml(k8s_client=k8s_cli, yaml_file=p1_yaml)
>    tic = time.time()
>    logger.info(f"Pod: p1 re-applied")
>
>    # 7. watch pod1 events and wait it to be scheduled
>    p1_running = False
>    for event in w.stream(v1.list_namespaced_pod,
>                          field_selector=f'metadata.name=p1',
>                          namespace="default",
>                          timeout_seconds=int(600)):
>        p1 = event['object']
>        if p1.status.phase == "Running":
>            p1_running = True
>            logger.info(f"Pod: {p1.metadata.name} scheduled on node {p1.spec.node_name}")
>            break
>
>    if not p1_running:
>        logger.error(f"Pod: {p1.metadata.name} is not scheduled in 600s")
>        assert False
>    else:
>        logger.info(f"Pod: {p1.metadata.name}, " + f"time between add and scheduled: {time.time() - tic}")
>
>
>```
>
>
>### Anything else we need to know?
>
>Tested on 1.30.2
>
>/sig bugs scheduling
>
>### Kubernetes version
>
><details>
>
>Tested on 1.30.2
>
></details>
>
>
>### Cloud provider
>
><details>
>
></details>
>
>
>### OS version
>
><details>
>
>```console
># On Linux:
>$ cat /etc/os-release
># paste output here
>$ uname -a
># paste output here
>
># On Windows:
>C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
># paste output here
>```
>
></details>
>
>
>### Install tools
>
><details>
>
></details>
>
>
>### Container runtime (CRI) and version (if applicable)
>
><details>
>
></details>
>
>
>### Related plugins (CNI, CSI, ...) and versions (if applicable)
>
><details>
>
></details>
>


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

If a SIG or subproject determines this is a relevant issue, they will accept it by applying the `triage/accepted` label and provide further guidance.

The `triage/accepted` label can be added by org members by writing `/triage accepted` in a comment.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>
/cc
hi @scheduler-tester ,This is the result I got from running a python script. I can't seem to reproduce it.:
```
09:51:18 - root - INFO - Pod: p1 added
09:51:18 - root - INFO - Pod: p1 scheduled on node node0
09:51:18 - root - INFO - Pod: p2 added
09:51:18 - root - INFO - Pod: p1, event: ADDED
09:51:18 - root - INFO - deletion_timestamp: None, status: Running
09:51:18 - root - INFO - Pod: p1, event: MODIFIED
09:51:18 - root - INFO - deletion_timestamp: 2024-08-16 01:51:48+00:00, status: Running
09:51:18 - root - INFO - Pod: p1 is terminating
09:51:18 - root - INFO - Pod: p3 scheduled on node p3
09:51:21 - root - INFO - Pod: p1 re-applied
09:51:21 - root - INFO - Pod: p1 scheduled on node node0
09:51:21 - root - INFO - Pod: p1, time between add and scheduled: 0.040593862533569336
```

kubectl get pod -o wide after execute:
```
NAME   READY   STATUS    RESTARTS   AGE     IP            NODE     NOMINATED NODE   READINESS GATES
p1     1/1     Running   0          2m18s   10.244.8.14   node0    <none>           <none>
p2     0/1     Pending   0          2m20s   <none>        <none>   <none>           <none>
p3     1/1     Running   0          2m20s   10.244.8.13   node0    <none>           <none>
```
@chengjoey  Hi, thanks for your swift reply! 
I'm sorry for my oversight: this bug only occurs without the 2-second sleep (as mentioned in the description above). Therefore, the `sleep(2)` between step 5 and step 6 in the python script should be commented out if you want to reproduce the bug (I've updated the script above to reflect this change).

Below is the log of running the script without `sleep(2)`:
```
10:10:10 - __main__ - INFO - Pod: p1 added
10:10:11 - __main__ - INFO - Pod: p1 scheduled on node node0
10:10:11 - __main__ - INFO - Pod: p2 added
10:10:11 - __main__ - INFO - Pod: p1, event: ADDED
10:10:11 - __main__ - INFO - deletion_timestamp: None, status: Running
10:10:11 - __main__ - INFO - Pod: p1, event: MODIFIED
10:10:11 - __main__ - INFO - deletion_timestamp: None, status: Pending
10:10:11 - __main__ - INFO - Pod: p1, event: MODIFIED
10:10:11 - __main__ - INFO - deletion_timestamp: 2024-08-16 02:10:41+00:00, status: Pending
10:10:11 - __main__ - INFO - Pod: p1 is terminating
10:10:11 - __main__ - INFO - Pod: p3 scheduled on node p3
10:10:11 - __main__ - INFO - Pod: p1 re-applied
10:15:18 - __main__ - INFO - Pod: p1 scheduled on node node0
10:15:18 - __main__ - INFO - Pod: p1, time between add and scheduled: 306.9531891345978
```

`kubectl get pod -o wide`: 1 min after execute
```
NAME   READY   STATUS    RESTARTS   AGE   IP         NODE     NOMINATED NODE   READINESS GATES
p1     0/1     Pending   0          59s   <none>     <none>   <none>           <none>
p2     0/1     Pending   0          59s   <none>     <none>   <none>           <none>
p3     1/1     Running   0          59s   10.0.0.3   node0    <none>           <none>

```
`kubectl get pod -o wide`: 5 minutes after execute
```
NAME   READY   STATUS    RESTARTS   AGE     IP         NODE     NOMINATED NODE   READINESS GATES
p1     1/1     Running   0          5m30s   10.0.0.4   node0    <none>           <none>
p2     0/1     Pending   0          5m30s   <none>     <none>   <none>           <none>
p3     1/1     Running   0          5m30s   10.0.0.3   node0    <none>           <none>
```


@scheduler-tester thanks, I reproduced it
hi @scheduler-tester , I found that it is not only because of `nominated_node_name`, but also related to priority. I think it cannot be simply fixed according to 2 in how to fix. Although p1 and p2 are both nominated, if the priority of p2 is not higher than p1, p1 can still be scheduled quickly. If I fix it according to `2`, I think it will destroy some existing logic

https://github.com/kubernetes/kubernetes/blob/c999f9d828009f0a535237aca74ce2e5f6a8088c/pkg/scheduler/framework/runtime/framework.go#L1035-L1045

If I comment out `nodeInfoOut.AddPodInfo(pi)`, p1 can be scheduled to node0 immediately

```
23:51:35 - root - INFO - Pod: p1 added
23:51:35 - root - INFO - Pod: p1 scheduled on node node0
23:51:35 - root - INFO - Pod: p2 added
23:51:35 - root - INFO - Pod: p1, event: ADDED
23:51:35 - root - INFO - deletion_timestamp: 2024-08-16 15:52:05+00:00, status: Running
23:51:35 - root - INFO - Pod: p1 is terminating
23:51:35 - root - INFO - Pod: p3 scheduled on node p3
23:51:35 - root - INFO - Pod: p1 re-applied
23:51:35 - root - INFO - Pod: p1 scheduled on node node0
23:51:35 - root - INFO - Pod: p1, time between add and scheduled: 0.014630317687988281
```
@scheduler-tester ,I also found an interesting phenomenon. Change p1 in step `4` to another pod `p4`. p4 and p1 need the same resource `10Gi`, but p4's priority is the same as p2's, both are `medium-priority`. p4 also needs to wait 5 minutes to be scheduled to node0.

p4:
```
# p4.yaml
apiVersion: v1
kind: Pod
metadata:
  name: p4
spec:
  containers:
  - name: p4-container
    image: nginx
    resources:
      requests:
        memory: "10Gi"
  # Note that it is medium-priority like p2
  priorityClassName: medium-priority
```

I think pods of the same priority should not assume that resources are already occupied by the nominated node, because they are all of the same priority. This is unfair to the later pods of the same priority, and the resources required by the later pods may be sufficient, but they are assumed to be occupied by the previous pods.

https://github.com/kubernetes/kubernetes/blob/c999f9d828009f0a535237aca74ce2e5f6a8088c/pkg/scheduler/framework/runtime/framework.go#L1035-L1036

Hi @chengjoey, thanks for your help!
>  If I fix it according to 2, I think it will destroy some existing logic

You're right that if we clean all the nominated pods some logic will be broken.
So maybe we can consider:
1) The most direct way to fix the root cause (where the pod's `nominated_node_name` is not cleaned when itself can no longer be bound to the nominated node) is: When a later pod was bound to a node with nominated pods, check each nominated pod individually. If a pod cannot be bound on the node after the current pod is bound, its `nominated_node_name` should be cleaned.
2) Alternatively, we can register the relevent plugins to a new `UpdatePodNominatedNodeName` event. I noticed there are existing events like `UpdatePodLabel` and `UpdatePodScaleDown`, so adding a new event for this fine-grained update might be helpful. However, I'm doubting that `nominated_node_name` is a field managed by the scheduler, not the user, so maybe there're better solution to do this (trigger the pod is retried as soon some `nominated_node_name` is cleared).

> If I comment out `nodeInfoOut.AddPodInfo(pi)`, p1 can be scheduled to node0 immediately

That's true, but I think this should be kept because according to [the comment](https://github.com/kubernetes/kubernetes/blob/c999f9d828009f0a535237aca74ce2e5f6a8088c/pkg/scheduler/framework/runtime/framework.go#L977C2-L977C82), the scheduler need two pass filter (one pass with nominated pods, one pass without) to make sure the decision is compatible with the nominated pods no matter they are scheduled to the nominated node or not. 

---

> I think pods of the same priority should not assume that resources are already occupied by the nominated node, because they are all of the same priority. 

In my opinion, this logic is sound. P1 not only arrives earlier but is also the one preempting other pods. Therefore, it seems reasonable to reserve some resources for it.

> This is unfair to the later pods of the same priority, and the resources required by the later pods may be sufficient, but they are assumed to be occupied by the previous pods.

I think this behavior is intentional (though I'm not sure), so removing it might disrupt existing logic. Maybe we should ask for member's idea on this.
Alternatively, this issue could be addressed by clearing the nominated_node_name of the previous pod, possibly via the fixes mentioned above.
@dom4ha @sanposhiho could this affect the discussions in #126858?
It's an interesting corner case.
So, the scheduler should regard the update event of Pods that clears the nominated node info as the Pod deletion event, but only effective for lower priority Pods. 

So, in this issue's example, when `p2`'s nominatedNodeName is cleared, we should regard the update event to `p2` as Pod/Delete, and move `p1` back to the activeQ. (The resource fit plugin, which should be registered as a unsched plugin of `p2` registers Pod/Delete event) 
And, let's say the scheduler also has `p4` that has the highest priority. `p4` shouldn't be requeued to activeQ. That's what I meant by "only effective for lower priority Pods".

> could this affect the discussions in https://github.com/kubernetes/kubernetes/issues/126858?

My current take is that it's a requeueing (event handling) problem, not a preemption problem, and hence I'd say it's not much related to that enhancement, for now. 
/cc @macsko 

Fix patch:
diff --git a/pkg/scheduler/eventhandlers.go b/pkg/scheduler/eventhandlers.go
index 3b009d76fb613..a6b1f066c77e6 100644
--- a/pkg/scheduler/eventhandlers.go
+++ b/pkg/scheduler/eventhandlers.go
@@ -160,6 +160,16 @@ func (sched *Scheduler) updatePodInSchedulingQueue(oldObj, newObj interface{}) {
 
 	logger.V(4).Info("Update event for unscheduled pod", "pod", klog.KObj(newPod))
 	sched.SchedulingQueue.Update(logger, oldPod, newPod)
+	if hasNominatedNodeNameChanged(oldPod, newPod) {
+		// Nominated node changed in pod, so we need to treat it as if the pod was deleted from the old nominated node,
+		// because the scheduler treats such a pod as if it was already assigned when scheduling lower or equal priority pods.
+		sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(logger, framework.EventAssignedPodDelete, oldPod, nil, getLEPriorityPreCheck(corev1helpers.PodPriority(oldPod)))
+	}
+}
+
+// hasNominatedNodeNameChanged returns true when nominated node name has existed but changed.
+func hasNominatedNodeNameChanged(oldPod, newPod *v1.Pod) bool {
+	return len(oldPod.Status.NominatedNodeName) > 0 && oldPod.Status.NominatedNodeName != newPod.Status.NominatedNodeName
 }
 
 func (sched *Scheduler) deletePodFromSchedulingQueue(obj interface{}) {
@@ -195,8 +205,21 @@ func (sched *Scheduler) deletePodFromSchedulingQueue(obj interface{}) {
 	// If a waiting pod is rejected, it indicates it's previously assumed and we're
 	// removing it from the scheduler cache. In this case, signal a AssignedPodDelete
 	// event to immediately retry some unscheduled Pods.
+	// Similarly when a pod that had nominated node is deleted, it can unblock scheduling of other pods,
+	// because the lower or equal priority pods treat such a pod as if it was assigned.
 	if fwk.RejectWaitingPod(pod.UID) {
 		sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(logger, framework.EventAssignedPodDelete, pod, nil, nil)
+	} else if pod.Status.NominatedNodeName != "" {
+		// Note that a nominated pod can fall into `RejectWaitingPod` case as well,
+		// but in that case the `MoveAllToActiveOrBackoffQueue` already covered lower priority pods.
+		sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(logger, framework.EventAssignedPodDelete, pod, nil, getLEPriorityPreCheck(corev1helpers.PodPriority(pod)))
+	}
+}
+
+// getLEPriorityPreCheck is a PreEnqueueCheck function that selects only lower or equal priority pods.
+func getLEPriorityPreCheck(priority int32) queue.PreEnqueueCheck {
+	return func(pod *v1.Pod) bool {
+		return corev1helpers.PodPriority(pod) <= priority
 	}
 }
 
diff --git a/pkg/scheduler/framework/plugins/nodeports/node_ports.go b/pkg/scheduler/framework/plugins/nodeports/node_ports.go
index 74350150c3c05..21835f6c122f8 100644
--- a/pkg/scheduler/framework/plugins/nodeports/node_ports.go
+++ b/pkg/scheduler/framework/plugins/nodeports/node_ports.go
@@ -143,7 +143,7 @@ func (pl *NodePorts) isSchedulableAfterPodDeleted(logger klog.Logger, pod *v1.Po
 	}
 
 	// If the deleted pod is unscheduled, it doesn't make the target pod schedulable.
-	if deletedPod.Spec.NodeName == "" {
+	if deletedPod.Spec.NodeName == "" && deletedPod.Status.NominatedNodeName == "" {
 		logger.V(4).Info("the deleted pod is unscheduled and it doesn't make the target pod schedulable", "pod", klog.KObj(pod), "deletedPod", klog.KObj(deletedPod))
 		return framework.QueueSkip, nil
 	}
diff --git a/pkg/scheduler/framework/plugins/noderesources/fit.go b/pkg/scheduler/framework/plugins/noderesources/fit.go
index ef6c2244c6b01..d94554ad6d73c 100644
--- a/pkg/scheduler/framework/plugins/noderesources/fit.go
+++ b/pkg/scheduler/framework/plugins/noderesources/fit.go
@@ -294,7 +294,7 @@ func (f *Fit) isSchedulableAfterPodEvent(logger klog.Logger, pod *v1.Pod, oldObj
 	}
 
 	if modifiedPod == nil {
-		if originalPod.Spec.NodeName == "" {
+		if originalPod.Spec.NodeName == "" && originalPod.Status.NominatedNodeName == "" {
 			logger.V(5).Info("the deleted pod was unscheduled and it wouldn't make the unscheduled pod schedulable", "pod", klog.KObj(pod), "deletedPod", klog.KObj(originalPod))
 			return framework.QueueSkip, nil
 		}
diff --git a/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go b/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go
index b0c35b4c6603d..2387b5ac85ab8 100644
--- a/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go
+++ b/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go
@@ -104,7 +104,7 @@ func (pl *CSILimits) isSchedulableAfterPodDeleted(logger klog.Logger, pod *v1.Po
 		return framework.QueueSkip, nil
 	}
 
-	if deletedPod.Spec.NodeName == "" {
+	if deletedPod.Spec.NodeName == "" && deletedPod.Status.NominatedNodeName == "" {
 		return framework.QueueSkip, nil
 	}
 
diff --git a/pkg/scheduler/framework/runtime/framework.go b/pkg/scheduler/framework/runtime/framework.go
index dbdc6d0986ef3..77dc23bc52d0a 100644
--- a/pkg/scheduler/framework/runtime/framework.go
+++ b/pkg/scheduler/framework/runtime/framework.go
@@ -1011,7 +1011,7 @@ func (f *frameworkImpl) RunFilterPluginsWithNominatedPods(ctx context.Context, s
 		nodeInfoToUse := info
 		if i == 0 {
 			var err error
-			podsAdded, stateToUse, nodeInfoToUse, err = addNominatedPods(ctx, f, pod, state, info)
+			podsAdded, stateToUse, nodeInfoToUse, err = addGENominatedPods(ctx, f, pod, state, info)
 			if err != nil {
 				return framework.AsStatus(err)
 			}
@@ -1028,10 +1028,10 @@ func (f *frameworkImpl) RunFilterPluginsWithNominatedPods(ctx context.Context, s
 	return status
 }
 
-// addNominatedPods adds pods with equal or greater priority which are nominated
+// addGENominatedPods adds pods with equal or greater priority which are nominated
 // to run on the node. It returns 1) whether any pod was added, 2) augmented cycleState,
 // 3) augmented nodeInfo.
-func addNominatedPods(ctx context.Context, fh framework.Handle, pod *v1.Pod, state *framework.CycleState, nodeInfo *framework.NodeInfo) (bool, *framework.CycleState, *framework.NodeInfo, error) {
+func addGENominatedPods(ctx context.Context, fh framework.Handle, pod *v1.Pod, state *framework.CycleState, nodeInfo *framework.NodeInfo) (bool, *framework.CycleState, *framework.NodeInfo, error) {
 	if fh == nil {
 		// This may happen only in tests.
 		return false, state, nodeInfo, nil
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index e0b6ba644dad1..857fa1af9e858 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -50,6 +50,7 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
 	"k8s.io/kubernetes/pkg/scheduler/profile"
 	"k8s.io/kubernetes/pkg/scheduler/util/assumecache"
+	"k8s.io/utils/clock"
 )
 
 const (
@@ -116,6 +117,7 @@ func (sched *Scheduler) applyDefaultHandlers() {
 }
 
 type schedulerOptions struct {
+	clock                  clock.Clock
 	componentConfigVersion string
 	kubeConfig             *restclient.Config
 	// Overridden by profile level percentageOfNodesToScore if set in v1.
@@ -227,6 +229,13 @@ func WithExtenders(e ...schedulerapi.Extender) Option {
 	}
 }
 
+// WithClock sets clock for PriorityQueue, the default clock is clock.RealClock.
+func WithClock(clock clock.Clock) Option {
+	return func(o *schedulerOptions) {
+		o.clock = clock
+	}
+}
+
 // FrameworkCapturer is used for registering a notify function in building framework.
 type FrameworkCapturer func(schedulerapi.KubeSchedulerProfile)
 
@@ -238,6 +247,7 @@ func WithBuildFrameworkCapturer(fc FrameworkCapturer) Option {
 }
 
 var defaultSchedulerOptions = schedulerOptions{
+	clock:                             clock.RealClock{},
 	percentageOfNodesToScore:          schedulerapi.DefaultPercentageOfNodesToScore,
 	podInitialBackoffSeconds:          int64(internalqueue.DefaultPodInitialBackoffDuration.Seconds()),
 	podMaxBackoffSeconds:              int64(internalqueue.DefaultPodMaxBackoffDuration.Seconds()),
@@ -343,6 +353,7 @@ func New(ctx context.Context,
 	podQueue := internalqueue.NewSchedulingQueue(
 		profiles[options.profiles[0].SchedulerName].QueueSortFunc(),
 		informerFactory,
+		internalqueue.WithClock(options.clock),
 		internalqueue.WithPodInitialBackoffDuration(time.Duration(options.podInitialBackoffSeconds)*time.Second),
 		internalqueue.WithPodMaxBackoffDuration(time.Duration(options.podMaxBackoffSeconds)*time.Second),
 		internalqueue.WithPodLister(podLister),


Test patch:
diff --git a/pkg/scheduler/eventhandlers_test.go b/pkg/scheduler/eventhandlers_test.go
index 37c7341a858e2..8f3f0b2134589 100644
--- a/pkg/scheduler/eventhandlers_test.go
+++ b/pkg/scheduler/eventhandlers_test.go
@@ -18,6 +18,7 @@ package scheduler
 
 import (
 	"context"
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -30,8 +31,10 @@ import (
 	resourceapi "k8s.io/api/resource/v1beta1"
 	storagev1 "k8s.io/api/storage/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/sets"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/klog/v2"
 	"k8s.io/klog/v2/ktesting"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -42,18 +45,168 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 
 	"k8s.io/kubernetes/pkg/features"
-	"k8s.io/kubernetes/pkg/scheduler/backend/cache"
-	"k8s.io/kubernetes/pkg/scheduler/backend/queue"
+	internalcache "k8s.io/kubernetes/pkg/scheduler/backend/cache"
+	internalqueue "k8s.io/kubernetes/pkg/scheduler/backend/queue"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodeaffinity"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodename"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodeports"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/noderesources"
+	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/queuesort"
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
+	"k8s.io/kubernetes/pkg/scheduler/util"
 	"k8s.io/kubernetes/pkg/scheduler/util/assumecache"
 )
 
+func TestEventHandlers_MoveToActiveOnNominatedNodeUpdate(t *testing.T) {
+	metrics.Register()
+	highPriorityPod :=
+		st.MakePod().Name("hpp").Namespace("ns1").UID("hppns1").Priority(highPriority).SchedulerName(testSchedulerName).Obj()
+
+	medNominatedPriorityPod :=
+		st.MakePod().Name("mpp").Namespace("ns2").UID("mppns1").Priority(midPriority).SchedulerName(testSchedulerName).NominatedNodeName("node1").Obj()
+	medPriorityPod :=
+		st.MakePod().Name("smpp").Namespace("ns3").UID("mppns2").Priority(midPriority).SchedulerName(testSchedulerName).Obj()
+
+	lowPriorityPod :=
+		st.MakePod().Name("lpp").Namespace("ns4").UID("lppns1").Priority(lowPriority).SchedulerName(testSchedulerName).Obj()
+
+	unschedulablePods := []*v1.Pod{highPriorityPod, medNominatedPriorityPod, medPriorityPod, lowPriorityPod}
+
+	// Make pods schedulable on Delete event when QHints are enabled, but not when nominated node appears.
+	queueHintForPodDelete := func(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (framework.QueueingHint, error) {
+		oldPod, _, err := util.As[*v1.Pod](oldObj, newObj)
+		if err != nil {
+			t.Errorf("Failed to convert objects to pods: %v", err)
+		}
+		if oldPod.Status.NominatedNodeName == "" {
+			return framework.QueueSkip, nil
+		}
+		return framework.Queue, nil
+	}
+	queueingHintMap := internalqueue.QueueingHintMapPerProfile{
+		testSchedulerName: {
+			framework.EventAssignedPodDelete: {
+				{
+					PluginName:     "fooPlugin1",
+					QueueingHintFn: queueHintForPodDelete,
+				},
+			},
+		},
+	}
+
+	tests := []struct {
+		name         string
+		updateFunc   func(s *Scheduler)
+		wantInActive sets.Set[string]
+	}{
+		{
+			name: "Update of a nominated node name to a different value should trigger rescheduling of lower priority pods",
+			updateFunc: func(s *Scheduler) {
+				updatedPod := medNominatedPriorityPod.DeepCopy()
+				updatedPod.Status.NominatedNodeName = "node2"
+				updatedPod.ResourceVersion = "1"
+				s.updatePodInSchedulingQueue(medNominatedPriorityPod, updatedPod)
+			},
+			wantInActive: sets.New(lowPriorityPod.Name, medPriorityPod.Name, medNominatedPriorityPod.Name),
+		},
+		{
+			name: "Removal of a nominated node name should trigger rescheduling of lower priority pods",
+			updateFunc: func(s *Scheduler) {
+				updatedPod := medNominatedPriorityPod.DeepCopy()
+				updatedPod.Status.NominatedNodeName = ""
+				updatedPod.ResourceVersion = "1"
+				s.updatePodInSchedulingQueue(medNominatedPriorityPod, updatedPod)
+			},
+			wantInActive: sets.New(lowPriorityPod.Name, medPriorityPod.Name, medNominatedPriorityPod.Name),
+		},
+		{
+			name: "Removal of a pod that had nominated node name should trigger rescheduling of lower priority pods",
+			updateFunc: func(s *Scheduler) {
+				s.deletePodFromSchedulingQueue(medNominatedPriorityPod)
+			},
+			wantInActive: sets.New(lowPriorityPod.Name, medPriorityPod.Name),
+		},
+		{
+			name: "Addition of a nominated node name to the high priority pod that did not have it before shouldn't trigger rescheduling",
+			updateFunc: func(s *Scheduler) {
+				updatedPod := highPriorityPod.DeepCopy()
+				updatedPod.Status.NominatedNodeName = "node2"
+				updatedPod.ResourceVersion = "1"
+				s.updatePodInSchedulingQueue(highPriorityPod, updatedPod)
+			},
+			wantInActive: sets.New[string](),
+		},
+	}
+
+	for _, tt := range tests {
+		for _, qHintEnabled := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s, with queuehint(%v)", tt.name, qHintEnabled), func(t *testing.T) {
+				featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerQueueingHints, qHintEnabled)
+
+				logger, ctx := ktesting.NewTestContext(t)
+				ctx, cancel := context.WithCancel(ctx)
+				defer cancel()
+
+				var objs []runtime.Object
+				for _, pod := range unschedulablePods {
+					objs = append(objs, pod)
+				}
+				client := fake.NewClientset(objs...)
+				informerFactory := informers.NewSharedInformerFactory(client, 0)
+
+				recorder := metrics.NewMetricsAsyncRecorder(3, 20*time.Microsecond, ctx.Done())
+				queue := internalqueue.NewPriorityQueue(
+					newDefaultQueueSort(),
+					informerFactory,
+					internalqueue.WithMetricsRecorder(*recorder),
+					internalqueue.WithQueueingHintMapPerProfile(queueingHintMap),
+					// disable backoff queue
+					internalqueue.WithPodInitialBackoffDuration(0),
+					internalqueue.WithPodMaxBackoffDuration(0))
+				schedulerCache := internalcache.New(ctx, 30*time.Second)
+
+				// Put test pods into unschedulable queue
+				for _, pod := range unschedulablePods {
+					queue.Add(logger, pod)
+					poppedPod, err := queue.Pop(logger)
+					if err != nil {
+						t.Fatalf("Pop failed: %v", err)
+					}
+					poppedPod.UnschedulablePlugins = sets.New("fooPlugin1")
+					if err := queue.AddUnschedulableIfNotPresent(logger, poppedPod, queue.SchedulingCycle()); err != nil {
+						t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err)
+					}
+				}
+
+				s, _, err := initScheduler(ctx, schedulerCache, queue, client, informerFactory)
+				if err != nil {
+					t.Fatalf("Failed to initialize test scheduler: %v", err)
+				}
+
+				if len(s.SchedulingQueue.PodsInActiveQ()) > 0 {
+					t.Errorf("No pods were expected to be in the activeQ before the update, but there were %v", s.SchedulingQueue.PodsInActiveQ())
+				}
+				tt.updateFunc(s)
+				if len(s.SchedulingQueue.PodsInActiveQ()) != len(tt.wantInActive) {
+					t.Errorf("Different number of pods were expected to be in the activeQ, but found actual %v vs. expected %v", s.SchedulingQueue.PodsInActiveQ(), tt.wantInActive)
+				}
+				for _, pod := range s.SchedulingQueue.PodsInActiveQ() {
+					if !tt.wantInActive.Has(pod.Name) {
+						t.Errorf("Found unexpected pod in activeQ: %s", pod.Name)
+					}
+				}
+			})
+		}
+	}
+}
+
+func newDefaultQueueSort() framework.LessFunc {
+	sort := &queuesort.PrioritySort{}
+	return sort.Less
+}
+
 func TestUpdatePodInCache(t *testing.T) {
 	ttl := 10 * time.Second
 	nodeName := "node"
@@ -81,8 +234,8 @@ func TestUpdatePodInCache(t *testing.T) {
 			ctx, cancel := context.WithCancel(ctx)
 			defer cancel()
 			sched := &Scheduler{
-				Cache:           cache.New(ctx, ttl),
-				SchedulingQueue: queue.NewTestQueue(ctx, nil),
+				Cache:           internalcache.New(ctx, ttl),
+				SchedulingQueue: internalqueue.NewTestQueue(ctx, nil),
 				logger:          logger,
 			}
 			sched.addPodToCache(tt.oldObj)
@@ -354,7 +507,7 @@ func TestAddAllEventHandlers(t *testing.T) {
 			defer cancel()
 
 			informerFactory := informers.NewSharedInformerFactory(fake.NewClientset(), 0)
-			schedulingQueue := queue.NewTestQueueWithInformerFactory(ctx, nil, informerFactory)
+			schedulingQueue := internalqueue.NewTestQueueWithInformerFactory(ctx, nil, informerFactory)
 			testSched := Scheduler{
 				StopEverything:  ctx.Done(),
 				SchedulingQueue: schedulingQueue,
diff --git a/test/integration/scheduler/eventhandler/eventhandler_test.go b/test/integration/scheduler/eventhandler/eventhandler_test.go
index 1cd7d53336d79..b87bb60015429 100644
--- a/test/integration/scheduler/eventhandler/eventhandler_test.go
+++ b/test/integration/scheduler/eventhandler/eventhandler_test.go
@@ -18,13 +18,19 @@ package eventhandler
 
 import (
 	"context"
+	"fmt"
 	"testing"
+	"time"
 
 	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"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/component-helpers/scheduling/corev1"
 	configv1 "k8s.io/kube-scheduler/config/v1"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler"
 	configtesting "k8s.io/kubernetes/pkg/scheduler/apis/config/testing"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
@@ -32,9 +38,12 @@ import (
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 	schedulerutils "k8s.io/kubernetes/test/integration/scheduler"
 	testutils "k8s.io/kubernetes/test/integration/util"
+	testingclock "k8s.io/utils/clock/testing"
 	"k8s.io/utils/ptr"
 )
 
+var lowPriority, mediumPriority, highPriority int32 = 100, 200, 300
+
 var _ framework.FilterPlugin = &fooPlugin{}
 
 type fooPlugin struct {
@@ -135,3 +144,136 @@ func TestUpdateNodeEvent(t *testing.T) {
 		t.Errorf("Pod %v was not scheduled: %v", pod.Name, err)
 	}
 }
+
+func TestUpdateNominatedNodeName(t *testing.T) {
+	fakeClock := testingclock.NewFakeClock(time.Now())
+	testBackoff := time.Minute
+	testContext := testutils.InitTestAPIServer(t, "test-event", nil)
+	capacity := map[v1.ResourceName]string{
+		v1.ResourceMemory: "32",
+	}
+	var cleanupPods []*v1.Pod
+
+	testNode := st.MakeNode().Name("node-0").Label("kubernetes.io/hostname", "node-0").Capacity(capacity).Obj()
+	// Note that the low priority pod that cannot fit with the mid priority, but can fit with the high priority one.
+	podLow := testutils.InitPausePod(&testutils.PausePodConfig{
+		Name:      "test-lp-pod",
+		Namespace: testContext.NS.Name,
+		Priority:  &lowPriority,
+		Resources: &v1.ResourceRequirements{Requests: v1.ResourceList{
+			v1.ResourceMemory: *resource.NewQuantity(20, resource.DecimalSI)},
+		}})
+	cleanupPods = append(cleanupPods, podLow)
+	podMidNominated := testutils.InitPausePod(&testutils.PausePodConfig{
+		Name:      "test-nominated-pod",
+		Namespace: testContext.NS.Name,
+		Priority:  &mediumPriority,
+		Resources: &v1.ResourceRequirements{Requests: v1.ResourceList{
+			v1.ResourceMemory: *resource.NewQuantity(25, resource.DecimalSI)},
+		}})
+	cleanupPods = append(cleanupPods, podMidNominated)
+	podHigh := testutils.InitPausePod(&testutils.PausePodConfig{
+		Name:      "test-hp-pod",
+		Namespace: testContext.NS.Name,
+		Priority:  &highPriority,
+		Resources: &v1.ResourceRequirements{Requests: v1.ResourceList{
+			v1.ResourceMemory: *resource.NewQuantity(10, resource.DecimalSI)},
+		}})
+	cleanupPods = append(cleanupPods, podHigh)
+
+	tests := []struct {
+		name       string
+		updateFunc func(testCtx *testutils.TestContext)
+	}{
+		{
+			name: "Preempt nominated pod",
+			updateFunc: func(testCtx *testutils.TestContext) {
+				// Create high-priority pod and wait until it's scheduled (unnominate mid-priority pod)
+				pod, err := testutils.CreatePausePod(testCtx.ClientSet, podHigh)
+				if err != nil {
+					t.Fatalf("Creating pod error: %v", err)
+				}
+				if err = testutils.WaitForPodToSchedule(testCtx.Ctx, testCtx.ClientSet, pod); err != nil {
+					t.Fatalf("Pod %v was not scheduled: %v", pod.Name, err)
+				}
+			},
+		},
+		{
+			name: "Remove nominated pod",
+			updateFunc: func(testCtx *testutils.TestContext) {
+				if err := testutils.DeletePod(testCtx.ClientSet, podMidNominated.Name, podMidNominated.Namespace); err != nil {
+					t.Fatalf("Deleting pod error: %v", err)
+				}
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		for _, qHintEnabled := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s, with queuehint(%v)", tt.name, qHintEnabled), func(t *testing.T) {
+				featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerQueueingHints, qHintEnabled)
+
+				testCtx, teardown := schedulerutils.InitTestSchedulerForFrameworkTest(t, testContext, 0, true,
+					scheduler.WithClock(fakeClock),
+					// UpdateFunc needs to be called when the nominated pod is still in the backoff queue, thus small, but non 0 value.
+					scheduler.WithPodInitialBackoffSeconds(int64(testBackoff.Seconds())),
+					scheduler.WithPodMaxBackoffSeconds(int64(testBackoff.Seconds())),
+				)
+				defer teardown()
+
+				_, err := testutils.CreateNode(testCtx.ClientSet, testNode)
+				if err != nil {
+					t.Fatalf("Creating node error: %v", err)
+				}
+
+				// Create initial low-priority pod and wait until it's scheduled.
+				pod, err := testutils.CreatePausePod(testCtx.ClientSet, podLow)
+				if err != nil {
+					t.Fatalf("Creating pod error: %v", err)
+				}
+				if err := testutils.WaitForPodToSchedule(testCtx.Ctx, testCtx.ClientSet, pod); err != nil {
+					t.Fatalf("Pod %v was not scheduled: %v", pod.Name, err)
+				}
+
+				// Create mid-priority pod and wait until it becomes nominated (preempt low-priority pod) and remain uschedulable.
+				pod, err = testutils.CreatePausePod(testCtx.ClientSet, podMidNominated)
+				if err != nil {
+					t.Fatalf("Creating pod error: %v", err)
+				}
+				if err := testutils.WaitForNominatedNodeName(testCtx.Ctx, testCtx.ClientSet, pod); err != nil {
+					t.Errorf("NominatedNodeName field was not set for pod %v: %v", pod.Name, err)
+				}
+				if err := testutils.WaitForPodUnschedulable(testCtx.Ctx, testCtx.ClientSet, pod); err != nil {
+					t.Errorf("Pod %v haven't become unschedulabe: %v", pod.Name, err)
+				}
+
+				// Remove the initial low-priority pod, which will move the nominated unschedulable pod back to the backoff queue.
+				if err := testutils.DeletePod(testCtx.ClientSet, podLow.Name, podLow.Namespace); err != nil {
+					t.Fatalf("Deleting pod error: %v", err)
+				}
+
+				// Create another low-priority pods which cannot be scheduled because the mid-priority pod is nominated on the node and the node doesn't have enough resource to have two pods both.
+				pod, err = testutils.CreatePausePod(testCtx.ClientSet, podLow)
+				if err != nil {
+					t.Fatalf("Creating pod error: %v", err)
+				}
+				if err := testutils.WaitForPodUnschedulable(testCtx.Ctx, testCtx.ClientSet, pod); err != nil {
+					t.Fatalf("Pod %v was not scheduled: %v", pod.Name, err)
+				}
+
+				// Update causing the nominated pod to be removed or to get its nominated node name removed, which should trigger scheduling of the low priority pod.
+				// Note that the update has to happen since the nominated pod is still in the backoffQ to actually test updates of nominated, but not bound yet pods.
+				tt.updateFunc(testCtx)
+
+				// Advance time by the maxPodBackoffSeconds to move low priority pod out of the backoff queue.
+				fakeClock.Step(testBackoff)
+
+				// Expect the low-priority pod is notified about unnominated mid-pririty pod and gets scheduled, as it should fit this time.
+				if err := testutils.WaitForPodToSchedule(testCtx.Ctx, testCtx.ClientSet, podLow); err != nil {
+					t.Fatalf("Pod %v was not scheduled: %v", podLow.Name, err)
+				}
+				testutils.CleanupPods(testCtx.Ctx, testCtx.ClientSet, t, cleanupPods)
+			})
+		}
+	}
+}
diff --git a/test/integration/scheduler/preemption/preemption_test.go b/test/integration/scheduler/preemption/preemption_test.go
index dc60edb10ce90..348f56a384b68 100644
--- a/test/integration/scheduler/preemption/preemption_test.go
+++ b/test/integration/scheduler/preemption/preemption_test.go
@@ -84,26 +84,6 @@ const filterPluginName = "filter-plugin"
 
 var lowPriority, mediumPriority, highPriority = int32(100), int32(200), int32(300)
 
-func waitForNominatedNodeNameWithTimeout(ctx context.Context, cs clientset.Interface, pod *v1.Pod, timeout time.Duration) error {
-	if err := wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, timeout, false, func(ctx context.Context) (bool, error) {
-		pod, err := cs.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{})
-		if err != nil {
-			return false, err
-		}
-		if len(pod.Status.NominatedNodeName) > 0 {
-			return true, nil
-		}
-		return false, err
-	}); err != nil {
-		return fmt.Errorf(".status.nominatedNodeName of Pod %v/%v did not get set: %v", pod.Namespace, pod.Name, err)
-	}
-	return nil
-}
-
-func waitForNominatedNodeName(ctx context.Context, cs clientset.Interface, pod *v1.Pod) error {
-	return waitForNominatedNodeNameWithTimeout(ctx, cs, pod, wait.ForeverTestTimeout)
-}
-
 const tokenFilterName = "token-filter"
 
 // tokenFilter is a fake plugin that implements PreFilter and Filter.
@@ -504,7 +484,7 @@ func TestPreemption(t *testing.T) {
 				}
 				// Also check that the preemptor pod gets the NominatedNodeName field set.
 				if len(test.preemptedPodIndexes) > 0 {
-					if err := waitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
+					if err := testutils.WaitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
 						t.Errorf("NominatedNodeName field was not set for pod %v: %v", preemptor.Name, err)
 					}
 				}
@@ -1086,7 +1066,7 @@ func TestNonPreemption(t *testing.T) {
 					t.Fatalf("Error while creating preemptor: %v", err)
 				}
 
-				err = waitForNominatedNodeNameWithTimeout(testCtx.Ctx, cs, preemptorPod, 5*time.Second)
+				err = testutils.WaitForNominatedNodeNameWithTimeout(testCtx.Ctx, cs, preemptorPod, 5*time.Second)
 				// test.PreemptionPolicy == nil means we expect the preemptor to be nominated.
 				expect := test.PreemptionPolicy == nil
 				// err == nil indicates the preemptor is indeed nominated.
@@ -1168,7 +1148,7 @@ func TestDisablePreemption(t *testing.T) {
 				}
 
 				// Ensure preemptor should not be nominated.
-				if err := waitForNominatedNodeNameWithTimeout(testCtx.Ctx, cs, preemptor, 5*time.Second); err == nil {
+				if err := testutils.WaitForNominatedNodeNameWithTimeout(testCtx.Ctx, cs, preemptor, 5*time.Second); err == nil {
 					t.Errorf("Preemptor %v should not be nominated", preemptor.Name)
 				}
 
@@ -1381,7 +1361,7 @@ func TestPreemptionStarvation(t *testing.T) {
 					t.Errorf("Error while creating the preempting pod: %v", err)
 				}
 				// Check if .status.nominatedNodeName of the preemptor pod gets set.
-				if err := waitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
+				if err := testutils.WaitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
 					t.Errorf(".status.nominatedNodeName was not set for pod %v/%v: %v", preemptor.Namespace, preemptor.Name, err)
 				}
 				// Make sure that preemptor is scheduled after preemptions.
@@ -1481,7 +1461,7 @@ func TestPreemptionRaces(t *testing.T) {
 						}
 					}
 					// Check that the preemptor pod gets nominated node name.
-					if err := waitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
+					if err := testutils.WaitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
 						t.Errorf(".status.nominatedNodeName was not set for pod %v/%v: %v", preemptor.Namespace, preemptor.Name, err)
 					}
 					// Make sure that preemptor is scheduled after preemptions.
@@ -1577,8 +1557,8 @@ func TestNominatedNodeCleanUp(t *testing.T) {
 			},
 			postChecks: []func(ctx context.Context, cs clientset.Interface, pod *v1.Pod) error{
 				testutils.WaitForPodToSchedule,
-				waitForNominatedNodeName,
-				waitForNominatedNodeName,
+				testutils.WaitForNominatedNodeName,
+				testutils.WaitForNominatedNodeName,
 			},
 		},
 		{
@@ -1597,7 +1577,7 @@ func TestNominatedNodeCleanUp(t *testing.T) {
 			},
 			postChecks: []func(ctx context.Context, cs clientset.Interface, pod *v1.Pod) error{
 				testutils.WaitForPodToSchedule,
-				waitForNominatedNodeName,
+				testutils.WaitForNominatedNodeName,
 				testutils.WaitForPodToSchedule,
 			},
 			podNamesToDelete: []string{"low"},
@@ -1615,7 +1595,7 @@ func TestNominatedNodeCleanUp(t *testing.T) {
 			},
 			postChecks: []func(ctx context.Context, cs clientset.Interface, pod *v1.Pod) error{
 				testutils.WaitForPodToSchedule,
-				waitForNominatedNodeName,
+				testutils.WaitForNominatedNodeName,
 			},
 			// Delete the node to simulate an ErrNoNodesAvailable error.
 			deleteNode:       true,
@@ -1634,7 +1614,7 @@ func TestNominatedNodeCleanUp(t *testing.T) {
 			},
 			postChecks: []func(ctx context.Context, cs clientset.Interface, pod *v1.Pod) error{
 				testutils.WaitForPodToSchedule,
-				waitForNominatedNodeName,
+				testutils.WaitForNominatedNodeName,
 			},
 			podNamesToDelete: []string{fmt.Sprintf("low-%v", doNotFailMe)},
 			customPlugins: &configv1.Plugins{
@@ -1990,7 +1970,7 @@ func TestPDBInPreemption(t *testing.T) {
 				}
 				// Also check if .status.nominatedNodeName of the preemptor pod gets set.
 				if len(test.preemptedPodIndexes) > 0 {
-					if err := waitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
+					if err := testutils.WaitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
 						t.Errorf("Test [%v]: .status.nominatedNodeName was not set for pod %v/%v: %v", test.name, preemptor.Namespace, preemptor.Name, err)
 					}
 				}
@@ -2483,7 +2463,7 @@ func TestReadWriteOncePodPreemption(t *testing.T) {
 				}
 				// Also check that the preemptor pod gets the NominatedNodeName field set.
 				if len(test.preemptedPodIndexes) > 0 {
-					if err := waitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
+					if err := testutils.WaitForNominatedNodeName(testCtx.Ctx, cs, preemptor); err != nil {
 						t.Errorf("NominatedNodeName field was not set for pod %v: %v", preemptor.Name, err)
 					}
 				}
diff --git a/test/integration/util/util.go b/test/integration/util/util.go
index 370b2010ccb02..72d7f65e513d3 100644
--- a/test/integration/util/util.go
+++ b/test/integration/util/util.go
@@ -1160,3 +1160,23 @@ func NextPodOrDie(t *testing.T, testCtx *TestContext) *schedulerframework.Queued
 	}
 	return podInfo
 }
+
+func WaitForNominatedNodeNameWithTimeout(ctx context.Context, cs clientset.Interface, pod *v1.Pod, timeout time.Duration) error {
+	if err := wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, timeout, false, func(ctx context.Context) (bool, error) {
+		pod, err := cs.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		if len(pod.Status.NominatedNodeName) > 0 {
+			return true, nil
+		}
+		return false, err
+	}); err != nil {
+		return fmt.Errorf(".status.nominatedNodeName of Pod %v/%v did not get set: %w", pod.Namespace, pod.Name, err)
+	}
+	return nil
+}
+
+func WaitForNominatedNodeName(ctx context.Context, cs clientset.Interface, pod *v1.Pod) error {
+	return WaitForNominatedNodeNameWithTimeout(ctx, cs, pod, wait.ForeverTestTimeout)
+}