Repository: kubernetes-sigs/scheduler-plugins
Pull request #855 (2025-01-13 12:24:06)

fix(QOSSort): refine queue sorting logic by adding inqueue timestamp
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide/first-contribution.md#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

#### What type of PR is this?
/kind bug
<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:
- After comparing with the k/k queuesort package, we seem to be missing timestamp comparison.
https://github.com/kubernetes/kubernetes/blob/35d6959ace07a4edc2a17da3bd4d0fcd625caa83/pkg/scheduler/framework/plugins/queuesort/priority_sort.go#L40-L49

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #None

#### Special notes for your reviewer:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
None
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/scheduler-plugins/pull/855#" title="Author self-approved">googs1025</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [ffromani](https://github.com/ffromani) for approval. For more information see [the Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fscheduler-plugins).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-sigs/scheduler-plugins/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["ffromani"]} -->
netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *kubernetes-sigs-scheduler-plugins* processing.


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 3c22204ca61e0af0d0c32b5a506ba9cf6d9676f2 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/kubernetes-sigs-scheduler-plugins/deploys/678505e9add9840008562f41 |
googs1025: friendly ping @Huang-Wei 
k8s-ci-robot: @googs1025: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-scheduler-plugins-integration-test-master | e30ae6802540043cc14daa7b3d4a040b3a1345e1 | [link](https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_scheduler-plugins/855/pull-scheduler-plugins-integration-test-master/1878787083355557888) | true | `/test pull-scheduler-plugins-integration-test-master`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=scheduler-plugins&pr=855). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Agoogs1025). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/scheduler-plugins/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @googs1025: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-scheduler-plugins-integration-test-master | c305dc7a93248c9065fc8e4b148db9c4ca4f688a | [link](https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_scheduler-plugins/855/pull-scheduler-plugins-integration-test-master/1878810117919477760) | true | `/test pull-scheduler-plugins-integration-test-master`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=scheduler-plugins&pr=855). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Agoogs1025). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/scheduler-plugins/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @googs1025: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-scheduler-plugins-integration-test-master | ab70fd56d3e3b8a1ffda9bcef4d278cb34ed7970 | [link](https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_scheduler-plugins/855/pull-scheduler-plugins-integration-test-master/1878818224099823616) | true | `/test pull-scheduler-plugins-integration-test-master`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=scheduler-plugins&pr=855). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Agoogs1025). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/scheduler-plugins/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
googs1025: /test pull-scheduler-plugins-integration-test-master
k8s-ci-robot: @googs1025: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-scheduler-plugins-integration-test-master | ab70fd56d3e3b8a1ffda9bcef4d278cb34ed7970 | [link](https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_scheduler-plugins/855/pull-scheduler-plugins-integration-test-master/1878824820469141504) | true | `/test pull-scheduler-plugins-integration-test-master`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=scheduler-plugins&pr=855). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Agoogs1025). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/scheduler-plugins/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/pkg/coscheduling/core/core.go b/pkg/coscheduling/core/core.go
index cd7ad6fb..17746761 100644
--- a/pkg/coscheduling/core/core.go
+++ b/pkg/coscheduling/core/core.go
@@ -28,10 +28,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/sets"
 	informerv1 "k8s.io/client-go/informers/core/v1"
 	listerv1 "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -66,11 +64,10 @@ func (s *PermitState) Clone() framework.StateData {
 type Manager interface {
 	PreFilter(context.Context, *corev1.Pod) error
 	Permit(context.Context, *framework.CycleState, *corev1.Pod) Status
-	Unreserve(context.Context, *corev1.Pod)
 	GetPodGroup(context.Context, *corev1.Pod) (string, *v1alpha1.PodGroup)
-	GetAssignedPodCount(string) int
 	GetCreationTimestamp(context.Context, *corev1.Pod, time.Time) time.Time
 	DeletePermittedPodGroup(context.Context, string)
+	CalculateAssignedPods(context.Context, string, string) int
 	ActivateSiblings(ctx context.Context, pod *corev1.Pod, state *framework.CycleState)
 	BackoffPodGroup(string, time.Duration)
 }
@@ -90,34 +87,9 @@ type PodGroupManager struct {
 	backedOffPG *gocache.Cache
 	// podLister is pod lister
 	podLister listerv1.PodLister
-	// assignedPodsByPG stores the pods assumed or bound for podgroups
-	assignedPodsByPG map[string]sets.Set[string]
 	sync.RWMutex
 }
 
-func AddPodFactory(pgMgr *PodGroupManager) func(obj interface{}) {
-	return func(obj interface{}) {
-		p, ok := obj.(*corev1.Pod)
-		if !ok {
-			return
-		}
-		if p.Spec.NodeName == "" {
-			return
-		}
-		pgFullName, _ := pgMgr.GetPodGroup(context.Background(), p)
-		if pgFullName == "" {
-			return
-		}
-		pgMgr.RWMutex.Lock()
-		defer pgMgr.RWMutex.Unlock()
-		if assigned, exist := pgMgr.assignedPodsByPG[pgFullName]; exist {
-			assigned.Insert(p.Name)
-		} else {
-			pgMgr.assignedPodsByPG[pgFullName] = sets.New(p.Name)
-		}
-	}
-}
-
 // NewPodGroupManager creates a new operation object.
 func NewPodGroupManager(client client.Client, snapshotSharedLister framework.SharedLister, scheduleTimeout *time.Duration, podInformer informerv1.PodInformer) *PodGroupManager {
 	pgMgr := &PodGroupManager{
@@ -127,43 +99,10 @@ func NewPodGroupManager(client client.Client, snapshotSharedLister framework.Sha
 		podLister:            podInformer.Lister(),
 		permittedPG:          gocache.New(3*time.Second, 3*time.Second),
 		backedOffPG:          gocache.New(10*time.Second, 10*time.Second),
-		assignedPodsByPG:     map[string]sets.Set[string]{},
 	}
-	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: AddPodFactory(pgMgr),
-		DeleteFunc: func(obj interface{}) {
-			switch t := obj.(type) {
-			case *corev1.Pod:
-				pod := t
-				if pod.Spec.NodeName == "" {
-					return
-				}
-				pgMgr.Unreserve(context.Background(), pod)
-				return
-			case cache.DeletedFinalStateUnknown:
-				pod, ok := t.Obj.(*corev1.Pod)
-				if !ok {
-					return
-				}
-				if pod.Spec.NodeName == "" {
-					return
-				}
-				pgMgr.Unreserve(context.Background(), pod)
-				return
-			default:
-				return
-			}
-		},
-	})
 	return pgMgr
 }
 
-func (pgMgr *PodGroupManager) GetAssignedPodCount(pgName string) int {
-	pgMgr.RWMutex.RLock()
-	defer pgMgr.RWMutex.RUnlock()
-	return len(pgMgr.assignedPodsByPG[pgName])
-}
-
 func (pgMgr *PodGroupManager) BackoffPodGroup(pgName string, backoff time.Duration) {
 	if backoff == time.Duration(0) {
 		return
@@ -283,23 +222,16 @@ func (pgMgr *PodGroupManager) Permit(ctx context.Context, state *framework.Cycle
 		return PodGroupNotFound
 	}
 
-	pgMgr.RWMutex.RLock()
-	defer pgMgr.RWMutex.RUnlock()
-	assigned, exist := pgMgr.assignedPodsByPG[pgFullName]
-	if !exist {
-		assigned = sets.Set[string]{}
-		pgMgr.assignedPodsByPG[pgFullName] = assigned
-	}
-	assigned.Insert(pod.Name)
+	assigned := pgMgr.CalculateAssignedPods(ctx, pg.Name, pg.Namespace)
 	// The number of pods that have been assigned nodes is calculated from the snapshot.
 	// The current pod in not included in the snapshot during the current scheduling cycle.
-	if len(assigned) >= int(pg.Spec.MinMember) {
+	if int32(assigned)+1 >= pg.Spec.MinMember {
 		return Success
 	}
 
-	if len(assigned) == 1 {
+	if assigned == 0 {
 		// Given we've reached Permit(), it's mean all PreFilter checks (minMember & minResource)
-		// already pass through, so if len(assigned) == 1, it could be due to:
+		// already pass through, so if assigned == 0, it could be due to:
 		// - minResource get satisfied
 		// - new pods added
 		// In either case, we should and only should use this 0-th pod to trigger activating
@@ -312,24 +244,6 @@ func (pgMgr *PodGroupManager) Permit(ctx context.Context, state *framework.Cycle
 	return Wait
 }
 
-// Unreserve invalidates assigned pod from assignedPodsByPG when schedule or bind failed.
-func (pgMgr *PodGroupManager) Unreserve(ctx context.Context, pod *corev1.Pod) {
-	pgFullName, _ := pgMgr.GetPodGroup(ctx, pod)
-	if pgFullName == "" {
-		return
-	}
-
-	pgMgr.RWMutex.Lock()
-	defer pgMgr.RWMutex.Unlock()
-	assigned, exist := pgMgr.assignedPodsByPG[pgFullName]
-	if exist {
-		assigned.Delete(pod.Name)
-		if len(assigned) == 0 {
-			delete(pgMgr.assignedPodsByPG, pgFullName)
-		}
-	}
-}
-
 // GetCreationTimestamp returns the creation time of a podGroup or a pod.
 func (pgMgr *PodGroupManager) GetCreationTimestamp(ctx context.Context, pod *corev1.Pod, ts time.Time) time.Time {
 	pgName := util.GetPodGroupLabel(pod)
@@ -361,6 +275,27 @@ func (pgMgr *PodGroupManager) GetPodGroup(ctx context.Context, pod *corev1.Pod)
 	return fmt.Sprintf("%v/%v", pod.Namespace, pgName), &pg
 }
 
+// CalculateAssignedPods returns the number of pods that has been assigned nodes: assumed or bound.
+func (pgMgr *PodGroupManager) CalculateAssignedPods(ctx context.Context, podGroupName, namespace string) int {
+	lh := klog.FromContext(ctx)
+	nodeInfos, err := pgMgr.snapshotSharedLister.NodeInfos().List()
+	if err != nil {
+		lh.Error(err, "Cannot get nodeInfos from frameworkHandle")
+		return 0
+	}
+	var count int
+	for _, nodeInfo := range nodeInfos {
+		for _, podInfo := range nodeInfo.Pods {
+			pod := podInfo.Pod
+			if util.GetPodGroupLabel(pod) == podGroupName && pod.Namespace == namespace && pod.Spec.NodeName != "" {
+				count++
+			}
+		}
+	}
+
+	return count
+}
+
 // CheckClusterResource checks if resource capacity of the cluster can satisfy <resourceRequest>.
 // It returns an error detailing the resource gap if not satisfied; otherwise returns nil.
 func CheckClusterResource(ctx context.Context, nodeList []*framework.NodeInfo, resourceRequest corev1.ResourceList, desiredPodGroupName string) error {
diff --git a/pkg/coscheduling/coscheduling.go b/pkg/coscheduling/coscheduling.go
index 1dcf8705..a6e50122 100644
--- a/pkg/coscheduling/coscheduling.go
+++ b/pkg/coscheduling/coscheduling.go
@@ -166,7 +166,7 @@ func (cs *Coscheduling) PostFilter(ctx context.Context, state *framework.CycleSt
 
 	// This indicates there are already enough Pods satisfying the PodGroup,
 	// so don't bother to reject the whole PodGroup.
-	assigned := cs.pgMgr.GetAssignedPodCount(pgName)
+	assigned := cs.pgMgr.CalculateAssignedPods(ctx, pg.Name, pod.Namespace)
 	if assigned >= int(pg.Spec.MinMember) {
 		lh.V(4).Info("Assigned pods", "podGroup", klog.KObj(pg), "assigned", assigned)
 		return &framework.PostFilterResult{}, framework.NewStatus(framework.Unschedulable)
@@ -256,7 +256,6 @@ func (cs *Coscheduling) Unreserve(ctx context.Context, state *framework.CycleSta
 	if pg == nil {
 		return
 	}
-	cs.pgMgr.Unreserve(ctx, pod)
 	cs.frameworkHandler.IterateOverWaitingPods(func(waitingPod framework.WaitingPod) {
 		if waitingPod.GetPod().Namespace == pod.Namespace && util.GetPodGroupLabel(waitingPod.GetPod()) == pg.Name {
 			lh.V(3).Info("Unreserve rejects", "pod", klog.KObj(waitingPod.GetPod()), "podGroup", klog.KObj(pg))
diff --git a/pkg/qos/queue_sort.go b/pkg/qos/queue_sort.go
index 8d7c3206..72ab9571 100644
--- a/pkg/qos/queue_sort.go
+++ b/pkg/qos/queue_sort.go
@@ -41,22 +41,43 @@ func (pl *Sort) Name() string {
 
 // Less is the function used by the activeQ heap algorithm to sort pods.
 // It sorts pods based on their priorities. When the priorities are equal, it uses
-// the Pod QoS classes to break the tie.
+// the Pod QoS classes to break the tie. If both the priority and QoS class are equal,
+// it uses PodQueueInfo.timestamp to determine the order.
 func (*Sort) Less(pInfo1, pInfo2 *framework.QueuedPodInfo) bool {
 	p1 := corev1helpers.PodPriority(pInfo1.Pod)
 	p2 := corev1helpers.PodPriority(pInfo2.Pod)
-	return (p1 > p2) || (p1 == p2 && compQOS(pInfo1.Pod, pInfo2.Pod))
+
+	if p1 != p2 {
+		return p1 > p2
+	}
+	qosResult := compQOS(pInfo1.Pod, pInfo2.Pod)
+	if qosResult != 0 {
+		return qosResult > 0
+	}
+	return pInfo1.Timestamp.Before(pInfo2.Timestamp)
 }
 
-func compQOS(p1, p2 *v1.Pod) bool {
+// compQOS compares the QoS classes of two Pods and returns:
+//
+//	 1 if p1 has a higher precedence QoS class than p2,
+//	-1 if p2 has a higher precedence QoS class than p1,
+//	 0 if both have the same QoS class.
+func compQOS(p1, p2 *v1.Pod) int {
 	p1QOS, p2QOS := v1qos.GetPodQOS(p1), v1qos.GetPodQOS(p2)
-	if p1QOS == v1.PodQOSGuaranteed {
-		return true
+
+	// Define the precedence order of QoS classes using a map
+	qosOrder := map[v1.PodQOSClass]int{
+		v1.PodQOSBestEffort: 1,
+		v1.PodQOSBurstable:  2,
+		v1.PodQOSGuaranteed: 3,
 	}
-	if p1QOS == v1.PodQOSBurstable {
-		return p2QOS != v1.PodQOSGuaranteed
+
+	if qosOrder[p1QOS] > qosOrder[p2QOS] {
+		return 1
+	} else if qosOrder[p1QOS] < qosOrder[p2QOS] {
+		return -1
 	}
-	return p2QOS == v1.PodQOSBestEffort
+	return 0
 }
 
 // New initializes a new plugin and returns it.


Test patch:
diff --git a/pkg/coscheduling/core/core_test.go b/pkg/coscheduling/core/core_test.go
index 0968492d..ba7205da 100644
--- a/pkg/coscheduling/core/core_test.go
+++ b/pkg/coscheduling/core/core_test.go
@@ -25,7 +25,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
 	clicache "k8s.io/client-go/tools/cache"
@@ -171,7 +170,6 @@ func TestPreFilter(t *testing.T) {
 				scheduleTimeout:      &scheduleTimeout,
 				permittedPG:          newCache(),
 				backedOffPG:          newCache(),
-				assignedPodsByPG:     make(map[string]sets.Set[string]),
 			}
 
 			informerFactory.Start(ctx.Done())
@@ -266,17 +264,19 @@ func TestPermit(t *testing.T) {
 			informerFactory := informers.NewSharedInformerFactory(cs, 0)
 			podInformer := informerFactory.Core().V1().Pods()
 
-			pgMgr := NewPodGroupManager(client, tu.NewFakeSharedLister(tt.existingPods, nodes), &scheduleTimeout, podInformer)
+			pgMgr := &PodGroupManager{
+				client:               client,
+				snapshotSharedLister: tu.NewFakeSharedLister(tt.existingPods, nodes),
+				podLister:            podInformer.Lister(),
+				scheduleTimeout:      &scheduleTimeout,
+			}
 
 			informerFactory.Start(ctx.Done())
 			if !clicache.WaitForCacheSync(ctx.Done(), podInformer.Informer().HasSynced) {
 				t.Fatal("WaitForCacheSync failed")
 			}
-			addFunc := AddPodFactory(pgMgr)
 			for _, p := range tt.existingPods {
 				podInformer.Informer().GetStore().Add(p)
-				// we call add func here because we can not ensure existing pods are added before premit are called
-				addFunc(p)
 			}
 
 			if got := pgMgr.Permit(ctx, &framework.CycleState{}, tt.pod); got != tt.want {
diff --git a/pkg/coscheduling/coscheduling_test.go b/pkg/coscheduling/coscheduling_test.go
index f0ade5f1..85cf73d5 100644
--- a/pkg/coscheduling/coscheduling_test.go
+++ b/pkg/coscheduling/coscheduling_test.go
@@ -622,27 +622,24 @@ func TestPostFilter(t *testing.T) {
 			cs := clientsetfake.NewSimpleClientset()
 			informerFactory := informers.NewSharedInformerFactory(cs, 0)
 			podInformer := informerFactory.Core().V1().Pods()
-			pgMgr := core.NewPodGroupManager(
-				client,
-				tu.NewFakeSharedLister(tt.existingPods, nodes),
-				&scheduleTimeout,
-				podInformer,
-			)
+
 			pl := &Coscheduling{
 				frameworkHandler: f,
-				pgMgr:            pgMgr,
-				scheduleTimeout:  &scheduleTimeout,
+				pgMgr: core.NewPodGroupManager(
+					client,
+					tu.NewFakeSharedLister(tt.existingPods, nodes),
+					&scheduleTimeout,
+					podInformer,
+				),
+				scheduleTimeout: &scheduleTimeout,
 			}
 
 			informerFactory.Start(ctx.Done())
 			if !clicache.WaitForCacheSync(ctx.Done(), podInformer.Informer().HasSynced) {
 				t.Fatal("WaitForCacheSync failed")
 			}
-			addFunc := core.AddPodFactory(pgMgr)
 			for _, p := range tt.existingPods {
 				podInformer.Informer().GetStore().Add(p)
-				// we call add func here because we can not ensure existing pods are added before premit are called
-				addFunc(p)
 			}
 
 			_, got := pl.PostFilter(ctx, framework.NewCycleState(), tt.pod, nodeStatusMap)
diff --git a/pkg/qos/queue_sort_test.go b/pkg/qos/queue_sort_test.go
index 4f3fe304..e97b19e4 100644
--- a/pkg/qos/queue_sort_test.go
+++ b/pkg/qos/queue_sort_test.go
@@ -18,6 +18,7 @@ package qos
 
 import (
 	"testing"
+	"time"
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -31,6 +32,8 @@ func createPodInfo(pod *v1.Pod) *framework.PodInfo {
 }
 
 func TestSortLess(t *testing.T) {
+	earlierTime := time.Now()
+	laterTime := earlierTime.Add(time.Second)
 	tests := []struct {
 		name   string
 		pInfo1 *framework.QueuedPodInfo
@@ -58,14 +61,16 @@ func TestSortLess(t *testing.T) {
 			want: false,
 		},
 		{
-			name: "p1 and p2 are both BestEfforts",
+			name: "p1 and p2 are both BestEfforts, but p2 is added to schedulingQ earlier than p1",
 			pInfo1: &framework.QueuedPodInfo{
-				PodInfo: createPodInfo(makePod("p1", 0, nil, nil)),
+				PodInfo:   createPodInfo(makePod("p1", 0, nil, nil)),
+				Timestamp: laterTime,
 			},
 			pInfo2: &framework.QueuedPodInfo{
-				PodInfo: createPodInfo(makePod("p2", 0, nil, nil)),
+				PodInfo:   createPodInfo(makePod("p2", 0, nil, nil)),
+				Timestamp: earlierTime,
 			},
-			want: true,
+			want: false,
 		},
 		{
 			name: "p1 is BestEfforts, p2 is Guaranteed",
@@ -88,14 +93,16 @@ func TestSortLess(t *testing.T) {
 			want: false,
 		},
 		{
-			name: "both p1 and p2 are Burstable",
+			name: "both p1 and p2 are Burstable, but p2 is added to schedulingQ earlier than p1",
 			pInfo1: &framework.QueuedPodInfo{
-				PodInfo: createPodInfo(makePod("p1", 0, getResList("100m", "100Mi"), getResList("200m", "200Mi"))),
+				PodInfo:   createPodInfo(makePod("p1", 0, getResList("100m", "100Mi"), getResList("200m", "200Mi"))),
+				Timestamp: laterTime,
 			},
 			pInfo2: &framework.QueuedPodInfo{
-				PodInfo: createPodInfo(makePod("p2", 0, getResList("100m", "100Mi"), getResList("200m", "200Mi"))),
+				PodInfo:   createPodInfo(makePod("p2", 0, getResList("100m", "100Mi"), getResList("200m", "200Mi"))),
+				Timestamp: earlierTime,
 			},
-			want: true,
+			want: false,
 		},
 		{
 			name: "p1 is Guaranteed, p2 is Burstable",
@@ -108,12 +115,26 @@ func TestSortLess(t *testing.T) {
 			want: true,
 		},
 		{
-			name: "both p1 and p2 are Guaranteed",
+			name: "both p1 and p2 are Guaranteed, but p1 is added to schedulingQ earlier than p2",
 			pInfo1: &framework.QueuedPodInfo{
-				PodInfo: createPodInfo(makePod("p1", 0, getResList("100m", "100Mi"), getResList("100m", "100Mi"))),
+				PodInfo:   createPodInfo(makePod("p1", 0, getResList("100m", "100Mi"), getResList("100m", "100Mi"))),
+				Timestamp: earlierTime,
 			},
 			pInfo2: &framework.QueuedPodInfo{
-				PodInfo: createPodInfo(makePod("p2", 0, getResList("100m", "100Mi"), getResList("100m", "100Mi"))),
+				PodInfo:   createPodInfo(makePod("p2", 0, getResList("100m", "100Mi"), getResList("100m", "100Mi"))),
+				Timestamp: laterTime,
+			},
+			want: true,
+		},
+		{
+			name: "both p1 and p2 are Guaranteed, but p1 is added to schedulingQ earlier than p2",
+			pInfo1: &framework.QueuedPodInfo{
+				PodInfo:   createPodInfo(makePod("p1", 0, getResList("100m", "100Mi"), getResList("100m", "100Mi"))),
+				Timestamp: earlierTime,
+			},
+			pInfo2: &framework.QueuedPodInfo{
+				PodInfo:   createPodInfo(makePod("p2", 0, getResList("100m", "100Mi"), getResList("100m", "100Mi"))),
+				Timestamp: laterTime,
 			},
 			want: true,
 		},
diff --git a/test/integration/coscheduling_test.go b/test/integration/coscheduling_test.go
index 18c514e8..a2bce1d3 100644
--- a/test/integration/coscheduling_test.go
+++ b/test/integration/coscheduling_test.go
@@ -376,138 +376,6 @@ func TestCoschedulingPlugin(t *testing.T) {
 	}
 }
 
-func TestPodCompleted(t *testing.T) {
-	testCtx := &testContext{}
-	testCtx.Ctx, testCtx.CancelFn = context.WithCancel(context.Background())
-
-	cs := kubernetes.NewForConfigOrDie(globalKubeConfig)
-	extClient := util.NewClientOrDie(globalKubeConfig)
-	testCtx.ClientSet = cs
-	testCtx.KubeConfig = globalKubeConfig
-
-	if err := wait.Poll(100*time.Millisecond, 3*time.Second, func() (done bool, err error) {
-		groupList, _, err := cs.ServerGroupsAndResources()
-		if err != nil {
-			return false, nil
-		}
-		for _, group := range groupList {
-			if group.Name == scheduling.GroupName {
-				t.Log("The CRD is ready to serve")
-				return true, nil
-			}
-		}
-		return false, nil
-	}); err != nil {
-		t.Fatalf("Timed out waiting for CRD to be ready: %v", err)
-	}
-
-	cfg, err := util.NewDefaultSchedulerComponentConfig()
-	if err != nil {
-		t.Fatal(err)
-	}
-	cfg.Profiles[0].Plugins.QueueSort = schedapi.PluginSet{
-		Enabled:  []schedapi.Plugin{{Name: coscheduling.Name}},
-		Disabled: []schedapi.Plugin{{Name: "*"}},
-	}
-	cfg.Profiles[0].Plugins.PreFilter.Enabled = append(cfg.Profiles[0].Plugins.PreFilter.Enabled, schedapi.Plugin{Name: coscheduling.Name})
-	cfg.Profiles[0].Plugins.PostFilter.Enabled = append(cfg.Profiles[0].Plugins.PostFilter.Enabled, schedapi.Plugin{Name: coscheduling.Name})
-	cfg.Profiles[0].Plugins.Permit.Enabled = append(cfg.Profiles[0].Plugins.Permit.Enabled, schedapi.Plugin{Name: coscheduling.Name})
-	cfg.Profiles[0].PluginConfig = append(cfg.Profiles[0].PluginConfig, schedapi.PluginConfig{
-		Name: coscheduling.Name,
-		Args: &schedconfig.CoschedulingArgs{
-			PermitWaitingTimeSeconds: 3,
-			PodGroupBackoffSeconds:   1,
-		},
-	})
-
-	ns := fmt.Sprintf("integration-test-%v", string(uuid.NewUUID()))
-	createNamespace(t, testCtx, ns)
-
-	testCtx = initTestSchedulerWithOptions(
-		t,
-		testCtx,
-		scheduler.WithProfiles(cfg.Profiles...),
-		scheduler.WithFrameworkOutOfTreeRegistry(fwkruntime.Registry{coscheduling.Name: coscheduling.New}),
-	)
-	syncInformerFactory(testCtx)
-	go testCtx.Scheduler.Run(testCtx.Ctx)
-	t.Log("Init scheduler success")
-	defer cleanupTest(t, testCtx)
-
-	// Create Nodes.
-	node1Name := "fake-node-1"
-	node1 := st.MakeNode().Name(node1Name).Label("node", node1Name).Obj()
-	node1.Status.Allocatable = v1.ResourceList{
-		v1.ResourceCPU:  *resource.NewQuantity(6, resource.DecimalSI),
-		v1.ResourcePods: *resource.NewQuantity(32, resource.DecimalSI),
-	}
-	node1.Status.Capacity = v1.ResourceList{
-		v1.ResourceCPU:  *resource.NewQuantity(6, resource.DecimalSI),
-		v1.ResourcePods: *resource.NewQuantity(32, resource.DecimalSI),
-	}
-	pause := imageutils.GetPauseImageName()
-	for _, tt := range []struct {
-		name      string
-		completed []*v1.Pod
-		pods      []*v1.Pod
-		podGroups []*v1alpha1.PodGroup
-	}{
-		{
-			name: "pg cannot be scheduled with completed pod",
-			completed: []*v1.Pod{
-				WithContainer(st.MakePod().Namespace(ns).Name("t1-p1-1").Req(map[v1.ResourceName]string{v1.ResourceCPU: "1"}).Priority(
-					midPriority).Label(v1alpha1.PodGroupLabel, "pg1-1").ZeroTerminationGracePeriod().Node(node1Name).Phase(v1.PodSucceeded).Obj(), pause)},
-			pods: []*v1.Pod{
-				WithContainer(st.MakePod().Namespace(ns).Name("t1-p1-2").Req(map[v1.ResourceName]string{v1.ResourceCPU: "1"}).Priority(
-					midPriority).Label(v1alpha1.PodGroupLabel, "pg1-1").ZeroTerminationGracePeriod().Obj(), pause),
-				WithContainer(st.MakePod().Namespace(ns).Name("t1-p1-3").Req(map[v1.ResourceName]string{v1.ResourceCPU: "1"}).Priority(
-					midPriority).Label(v1alpha1.PodGroupLabel, "pg1-1").ZeroTerminationGracePeriod().Obj(), pause),
-				WithContainer(st.MakePod().Namespace(ns).Name("t1-p1-4").Req(map[v1.ResourceName]string{v1.ResourceCPU: "1"}).Priority(
-					midPriority).Label(v1alpha1.PodGroupLabel, "pg1-1").ZeroTerminationGracePeriod().Obj(), pause),
-				WithContainer(st.MakePod().Namespace(ns).Name("t1-p1-5").Req(map[v1.ResourceName]string{v1.ResourceCPU: "1"}).Priority(
-					midPriority).Label(v1alpha1.PodGroupLabel, "pg1-1").ZeroTerminationGracePeriod().Obj(), pause),
-				WithContainer(st.MakePod().Namespace(ns).Name("t1-p1-6").Req(map[v1.ResourceName]string{v1.ResourceCPU: "1"}).Priority(
-					midPriority).Label(v1alpha1.PodGroupLabel, "pg1-1").ZeroTerminationGracePeriod().Obj(), pause),
-			},
-			podGroups: []*v1alpha1.PodGroup{
-				util.MakePG("pg1-1", ns, 6, nil, nil),
-			},
-		},
-	} {
-		t.Run(tt.name, func(t *testing.T) {
-			t.Logf("Start-coscheduling-test %v", tt.name)
-			defer cleanupPodGroups(testCtx.Ctx, extClient, tt.podGroups)
-			// create pod group
-			if err := createPodGroups(testCtx.Ctx, extClient, tt.podGroups); err != nil {
-				t.Fatal(err)
-			}
-			defer cleanupPods(t, testCtx, tt.pods)
-			// Create Pods, we will expect them to be scheduled in a reversed order.
-			for i := range tt.completed {
-				t.Logf("Creating pod %v", tt.completed[i].Name)
-				if _, err := cs.CoreV1().Pods(tt.completed[i].Namespace).Create(testCtx.Ctx, tt.completed[i], metav1.CreateOptions{}); err != nil {
-					t.Fatalf("Failed to create Pod %q: %v", tt.completed[i].Name, err)
-				}
-				t.Logf("Create pod %v success", tt.completed[i].Name)
-			}
-			for i := range tt.pods {
-				t.Logf("Creating pod %v", tt.pods[i].Name)
-				if _, err := cs.CoreV1().Pods(tt.pods[i].Namespace).Create(testCtx.Ctx, tt.pods[i], metav1.CreateOptions{}); err != nil {
-					t.Fatalf("Failed to create Pod %q: %v", tt.pods[i].Name, err)
-				}
-				t.Logf("Create pod %v success", tt.pods[i].Name)
-			}
-			time.Sleep(10 * time.Second)
-			for _, v := range tt.pods {
-				if podScheduled(t, cs, ns, v.Name) {
-					t.Fatalf("%v should not be scheduled", v.Name)
-				}
-			}
-			t.Logf("Case %v finished", tt.name)
-		})
-	}
-}
-
 func TestPodgroupBackoff(t *testing.T) {
 	testCtx := &testContext{}
 	testCtx.Ctx, testCtx.CancelFn = context.WithCancel(context.Background())
diff --git a/test/integration/qos_test.go b/test/integration/qos_test.go
index d78a1a2d..9fbea9b0 100644
--- a/test/integration/qos_test.go
+++ b/test/integration/qos_test.go
@@ -19,6 +19,9 @@ package integration
 import (
 	"context"
 	"fmt"
+	"reflect"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -114,19 +117,33 @@ func TestQOSPlugin(t *testing.T) {
 	// Create 3 Pods with the order: BestEfforts, Burstable, Guaranteed.
 	// We will expect them to be scheduled in a reversed order.
 	t.Logf("Start to create 3 Pods.")
-	for i := range pods {
-		t.Logf("Creating Pod %q", pods[i].Name)
-		_, err = cs.CoreV1().Pods(ns).Create(testCtx.Ctx, pods[i], metav1.CreateOptions{})
-		if err != nil {
-			t.Fatalf("Failed to create Pod %q: %v", pods[i].Name, err)
-		}
+	// Concurrently create all Pods.
+	var wg sync.WaitGroup
+	for _, pod := range pods {
+		wg.Add(1)
+		go func(p *v1.Pod) {
+			defer wg.Done()
+			_, err = cs.CoreV1().Pods(ns).Create(testCtx.Ctx, p, metav1.CreateOptions{})
+			if err != nil {
+				t.Errorf("Failed to create Pod %q: %v", p.Name, err)
+			} else {
+				t.Logf("Created Pod %q", p.Name)
+			}
+		}(pod)
 	}
+	wg.Wait()
 	defer cleanupPods(t, testCtx, pods)
 
 	// Wait for all Pods are in the scheduling queue.
 	err = wait.PollUntilContextTimeout(testCtx.Ctx, time.Millisecond*200, wait.ForeverTestTimeout, false, func(ctx context.Context) (bool, error) {
 		pendingPods, _ := testCtx.Scheduler.SchedulingQueue.PendingPods()
 		if len(pendingPods) == len(pods) {
+			// Collect Pod names into a slice.
+			podNames := make([]string, len(pendingPods))
+			for i, podInfo := range pendingPods {
+				podNames[i] = podInfo.Name
+			}
+			t.Logf("All Pods are in the pending queue: %v", strings.Join(podNames, ", "))
 			return true, nil
 		}
 		return false, nil
@@ -137,12 +154,17 @@ func TestQOSPlugin(t *testing.T) {
 
 	// Expect Pods are popped in the QoS class order.
 	logger := klog.FromContext(testCtx.Ctx)
-	for i := len(podNames) - 1; i >= 0; i-- {
+	expectedOrder := []string{"guaranteed", "burstable", "bestefforts"}
+	actualOrder := make([]string, len(expectedOrder))
+	for i := 0; i < len(expectedOrder); i++ {
 		podInfo, _ := testCtx.Scheduler.NextPod(logger)
-		if podInfo.Pod.Name != podNames[i] {
-			t.Errorf("Expect Pod %q, but got %q", podNames[i], podInfo.Pod.Name)
-		} else {
-			t.Logf("Pod %q is popped out as expected.", podInfo.Pod.Name)
-		}
+		actualOrder[i] = podInfo.Pod.Name
+		t.Logf("Popped Pod %q", podInfo.Pod.Name)
+	}
+
+	if !reflect.DeepEqual(actualOrder, expectedOrder) {
+		t.Errorf("Expected Pod order %v, but got %v", expectedOrder, actualOrder)
+	} else {
+		t.Logf("Pods were popped out in the expected order.")
 	}
 }