Repository: kubernetes-sigs/kueue

Issue:
Flaky Test: Topology Aware Scheduling when Delete Topology when ResourceFlavor exist should not allow to delete topology
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

If the matter is security related, please disclose it privately via https://kubernetes.io/security/
-->


**What happened**:
Failed `TopologyAwareScheduling Suite: [It] Topology Aware Scheduling when Delete Topology when ResourceFlavor exist should not allow to delete topology` in periodic CI.

```shell
{Failed after 0.258s.
The function passed to Consistently failed at /home/prow/go/src/kubernetes-sigs/kueue/test/integration/singlecluster/tas/tas_test.go:116 with:
Expected success, but got an error:
    <*errors.StatusError | 0xc000d806e0>: 
    topologies.kueue.x-k8s.io "topology" not found
    {
        ErrStatus: {
            TypeMeta: {Kind: "", APIVersion: ""},
            ListMeta: {
                SelfLink: "",
                ResourceVersion: "",
                Continue: "",
                RemainingItemCount: nil,
            },
            Status: "Failure",
            Message: "topologies.kueue.x-k8s.io \"topology\" not found",
            Reason: "NotFound",
            Details: {
                Name: "topology",
                Group: "kueue.x-k8s.io",
                Kind: "topologies",
                UID: "",
                Causes: nil,
                RetryAfterSeconds: 0,
            },
            Code: 404,
        },
    } failed [FAILED] Failed after 0.258s.
The function passed to Consistently failed at /home/prow/go/src/kubernetes-sigs/kueue/test/integration/singlecluster/tas/tas_test.go:116 with:
Expected success, but got an error:
    <*errors.StatusError | 0xc000d806e0>: 
    topologies.kueue.x-k8s.io "topology" not found
    {
        ErrStatus: {
            TypeMeta: {Kind: "", APIVersion: ""},
            ListMeta: {
                SelfLink: "",
                ResourceVersion: "",
                Continue: "",
                RemainingItemCount: nil,
            },
            Status: "Failure",
            Message: "topologies.kueue.x-k8s.io \"topology\" not found",
            Reason: "NotFound",
            Details: {
                Name: "topology",
                Group: "kueue.x-k8s.io",
                Kind: "topologies",
                UID: "",
                Causes: nil,
                RetryAfterSeconds: 0,
            },
            Code: 404,
        },
    }
In [It] at: /home/prow/go/src/kubernetes-sigs/kueue/test/integration/singlecluster/tas/tas_test.go:118 @ 03/05/25 18:31:18.676
}
```

**What you expected to happen**:
No errors.

**How to reproduce it (as minimally and precisely as possible)**:

https://prow.k8s.io/view/gs/kubernetes-ci-logs/logs/periodic-kueue-test-integration-main/1897350448038285312

<img width="1315" alt="Image" src="https://github.com/user-attachments/assets/40f53d21-e068-44a9-bf4c-210243c217f6" />

**Anything else we need to know?**:

**Environment**:
- Kubernetes version (use `kubectl version`):
- Kueue version (use `git describe --tags --dirty --always`):
- Cloud provider or hardware configuration:
- OS (e.g: `cat /etc/os-release`):
- Kernel (e.g. `uname -a`):
- Install tools:
- Others:

Hints from issue comments:
@tenzen-y: /kind flake
@mbobrovskyi: /assign

Solution patch:
diff --git a/apis/kueue/v1beta1/clusterqueue_types.go b/apis/kueue/v1beta1/clusterqueue_types.go
index 501555d85..0a5b9cd84 100644
--- a/apis/kueue/v1beta1/clusterqueue_types.go
+++ b/apis/kueue/v1beta1/clusterqueue_types.go
@@ -97,21 +97,6 @@ type ClusterQueueSpec struct {
 	// +kubebuilder:default={}
 	FlavorFungibility *FlavorFungibility `json:"flavorFungibility,omitempty"`
 
-	// preemption describes policies to preempt Workloads from this ClusterQueue
-	// or the ClusterQueue's cohort.
-	//
-	// Preemption can happen in two scenarios:
-	//
-	// - When a Workload fits within the nominal quota of the ClusterQueue, but
-	//   the quota is currently borrowed by other ClusterQueues in the cohort.
-	//   Preempting Workloads in other ClusterQueues allows this ClusterQueue to
-	//   reclaim its nominal quota.
-	// - When a Workload doesn't fit within the nominal quota of the ClusterQueue
-	//   and there are admitted Workloads in the ClusterQueue with lower priority.
-	//
-	// The preemption algorithm tries to find a minimal set of Workloads to
-	// preempt to accommodate the pending Workload, preempting Workloads with
-	// lower priority first.
 	// +kubebuilder:default={}
 	Preemption *ClusterQueuePreemption `json:"preemption,omitempty"`
 
@@ -414,6 +399,24 @@ type FlavorFungibility struct {
 
 // ClusterQueuePreemption contains policies to preempt Workloads from this
 // ClusterQueue or the ClusterQueue's cohort.
+//
+// Preemption may be configured to work in the following scenarios:
+//
+//   - When a Workload fits within the nominal quota of the ClusterQueue, but
+//     the quota is currently borrowed by other ClusterQueues in the cohort.
+//     We preempt workloads in other ClusterQueues to allow this ClusterQueue to
+//     reclaim its nominal quota. Configured using reclaimWithinCohort.
+//   - When a Workload doesn't fit within the nominal quota of the ClusterQueue
+//     and there are admitted Workloads in the ClusterQueue with lower priority.
+//     Configured using withinClusterQueue.
+//   - When a Workload may fit while both borrowing and preempting
+//     low priority workloads in the Cohort. Configured using borrowWithinCohort.
+//   - When FairSharing is enabled, to maintain fair distribution of
+//     unused resources. See FairSharing documentation.
+//
+// The preemption algorithm tries to find a minimal set of Workloads to
+// preempt to accomomdate the pending Workload, preempting Workloads with
+// lower priority first.
 // +kubebuilder:validation:XValidation:rule="!(self.reclaimWithinCohort == 'Never' && has(self.borrowWithinCohort) &&  self.borrowWithinCohort.policy != 'Never')", message="reclaimWithinCohort=Never and borrowWithinCohort.Policy!=Never"
 type ClusterQueuePreemption struct {
 	// reclaimWithinCohort determines whether a pending Workload can preempt
@@ -436,8 +439,6 @@ type ClusterQueuePreemption struct {
 	// +kubebuilder:validation:Enum=Never;LowerPriority;Any
 	ReclaimWithinCohort PreemptionPolicy `json:"reclaimWithinCohort,omitempty"`
 
-	// borrowWithinCohort provides configuration to allow preemption within
-	// cohort while borrowing.
 	// +kubebuilder:default={}
 	BorrowWithinCohort *BorrowWithinCohort `json:"borrowWithinCohort,omitempty"`
 
@@ -465,7 +466,8 @@ const (
 )
 
 // BorrowWithinCohort contains configuration which allows to preempt workloads
-// within cohort while borrowing.
+// within cohort while borrowing. It only works with Classical Preemption,
+// __not__ with Fair Sharing.
 type BorrowWithinCohort struct {
 	// policy determines the policy for preemption to reclaim quota within cohort while borrowing.
 	// Possible values are:
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
index 9ce77e2b9..8c897921b 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
@@ -245,27 +245,33 @@ spec:
               preemption:
                 default: {}
                 description: |-
-                  preemption describes policies to preempt Workloads from this ClusterQueue
-                  or the ClusterQueue's cohort.
+                  ClusterQueuePreemption contains policies to preempt Workloads from this
+                  ClusterQueue or the ClusterQueue's cohort.
 
-                  Preemption can happen in two scenarios:
+                  Preemption may be configured to work in the following scenarios:
 
-                  - When a Workload fits within the nominal quota of the ClusterQueue, but
-                    the quota is currently borrowed by other ClusterQueues in the cohort.
-                    Preempting Workloads in other ClusterQueues allows this ClusterQueue to
-                    reclaim its nominal quota.
-                  - When a Workload doesn't fit within the nominal quota of the ClusterQueue
-                    and there are admitted Workloads in the ClusterQueue with lower priority.
+                    - When a Workload fits within the nominal quota of the ClusterQueue, but
+                      the quota is currently borrowed by other ClusterQueues in the cohort.
+                      We preempt workloads in other ClusterQueues to allow this ClusterQueue to
+                      reclaim its nominal quota. Configured using reclaimWithinCohort.
+                    - When a Workload doesn't fit within the nominal quota of the ClusterQueue
+                      and there are admitted Workloads in the ClusterQueue with lower priority.
+                      Configured using withinClusterQueue.
+                    - When a Workload may fit while both borrowing and preempting
+                      low priority workloads in the Cohort. Configured using borrowWithinCohort.
+                    - When FairSharing is enabled, to maintain fair distribution of
+                      unused resources. See FairSharing documentation.
 
                   The preemption algorithm tries to find a minimal set of Workloads to
-                  preempt to accommodate the pending Workload, preempting Workloads with
+                  preempt to accomomdate the pending Workload, preempting Workloads with
                   lower priority first.
                 properties:
                   borrowWithinCohort:
                     default: {}
                     description: |-
-                      borrowWithinCohort provides configuration to allow preemption within
-                      cohort while borrowing.
+                      BorrowWithinCohort contains configuration which allows to preempt workloads
+                      within cohort while borrowing. It only works with Classical Preemption,
+                      __not__ with Fair Sharing.
                     properties:
                       maxPriorityThreshold:
                         description: |-
diff --git a/charts/kueue/templates/webhook/webhook.yaml b/charts/kueue/templates/webhook/webhook.yaml
index 4fb5d2e18..17179bf72 100644
--- a/charts/kueue/templates/webhook/webhook.yaml
+++ b/charts/kueue/templates/webhook/webhook.yaml
@@ -31,40 +31,6 @@ webhooks:
         resources:
           - appwrappers
     sideEffects: None
-  - admissionReviewVersions:
-      - v1
-    clientConfig:
-      service:
-        name: '{{ include "kueue.fullname" . }}-webhook-service'
-        namespace: '{{ .Release.Namespace }}'
-        path: /mutate--v1-pod
-    {{- if has "pod" $integrationsConfig.frameworks }}
-    failurePolicy: Fail
-    {{- else }}
-    failurePolicy: Ignore
-    {{- end }}
-    name: mpod.kb.io
-    namespaceSelector:
-      {{- if (hasKey $managerConfig "managedJobsNamespaceSelector") -}}
-        {{- toYaml $managerConfig.managedJobsNamespaceSelector | nindent 6 -}}
-      {{- else }}
-      matchExpressions:
-        - key: kubernetes.io/metadata.name
-          operator: NotIn
-          values:
-            - kube-system
-            - '{{ .Release.Namespace }}'
-      {{- end }}
-    rules:
-      - apiGroups:
-          - ""
-        apiVersions:
-          - v1
-        operations:
-          - CREATE
-        resources:
-          - pods
-    sideEffects: None
   - admissionReviewVersions:
       - v1
     clientConfig:
@@ -214,6 +180,40 @@ webhooks:
         resources:
           - xgboostjobs
     sideEffects: None
+  - admissionReviewVersions:
+      - v1
+    clientConfig:
+      service:
+        name: '{{ include "kueue.fullname" . }}-webhook-service'
+        namespace: '{{ .Release.Namespace }}'
+        path: /mutate--v1-pod
+    {{- if has "pod" $integrationsConfig.frameworks }}
+    failurePolicy: Fail
+    {{- else }}
+    failurePolicy: Ignore
+    {{- end }}
+    name: mpod.kb.io
+    namespaceSelector:
+      {{- if (hasKey $managerConfig "managedJobsNamespaceSelector") -}}
+        {{- toYaml $managerConfig.managedJobsNamespaceSelector | nindent 6 -}}
+      {{- else }}
+      matchExpressions:
+        - key: kubernetes.io/metadata.name
+          operator: NotIn
+          values:
+            - kube-system
+            - '{{ .Release.Namespace }}'
+      {{- end }}
+    rules:
+      - apiGroups:
+          - ""
+        apiVersions:
+          - v1
+        operations:
+          - CREATE
+        resources:
+          - pods
+    sideEffects: None
   - admissionReviewVersions:
       - v1
     clientConfig:
@@ -416,41 +416,6 @@ webhooks:
         resources:
           - appwrappers
     sideEffects: None
-  - admissionReviewVersions:
-      - v1
-    clientConfig:
-      service:
-        name: '{{ include "kueue.fullname" . }}-webhook-service'
-        namespace: '{{ .Release.Namespace }}'
-        path: /validate--v1-pod
-    {{- if has "pod" $integrationsConfig.frameworks }}
-    failurePolicy: Fail
-    {{- else }}
-    failurePolicy: Ignore
-    {{- end }}
-    name: vpod.kb.io
-    namespaceSelector:
-      {{- if (hasKey $managerConfig "managedJobsNamespaceSelector") -}}
-        {{- toYaml $managerConfig.managedJobsNamespaceSelector | nindent 6 -}}
-      {{- else }}
-      matchExpressions:
-        - key: kubernetes.io/metadata.name
-          operator: NotIn
-          values:
-            - kube-system
-            - '{{ .Release.Namespace }}'
-      {{- end }}
-    rules:
-      - apiGroups:
-          - ""
-        apiVersions:
-          - v1
-        operations:
-          - CREATE
-          - UPDATE
-        resources:
-          - pods
-    sideEffects: None
   - admissionReviewVersions:
       - v1
     clientConfig:
@@ -606,6 +571,41 @@ webhooks:
         resources:
           - xgboostjobs
     sideEffects: None
+  - admissionReviewVersions:
+      - v1
+    clientConfig:
+      service:
+        name: '{{ include "kueue.fullname" . }}-webhook-service'
+        namespace: '{{ .Release.Namespace }}'
+        path: /validate--v1-pod
+    {{- if has "pod" $integrationsConfig.frameworks }}
+    failurePolicy: Fail
+    {{- else }}
+    failurePolicy: Ignore
+    {{- end }}
+    name: vpod.kb.io
+    namespaceSelector:
+      {{- if (hasKey $managerConfig "managedJobsNamespaceSelector") -}}
+        {{- toYaml $managerConfig.managedJobsNamespaceSelector | nindent 6 -}}
+      {{- else }}
+      matchExpressions:
+        - key: kubernetes.io/metadata.name
+          operator: NotIn
+          values:
+            - kube-system
+            - '{{ .Release.Namespace }}'
+      {{- end }}
+    rules:
+      - apiGroups:
+          - ""
+        apiVersions:
+          - v1
+        operations:
+          - CREATE
+          - UPDATE
+        resources:
+          - pods
+    sideEffects: None
   - admissionReviewVersions:
       - v1
     clientConfig:
diff --git a/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml b/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
index 5e698e486..c1d9ea011 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
@@ -230,27 +230,33 @@ spec:
               preemption:
                 default: {}
                 description: |-
-                  preemption describes policies to preempt Workloads from this ClusterQueue
-                  or the ClusterQueue's cohort.
+                  ClusterQueuePreemption contains policies to preempt Workloads from this
+                  ClusterQueue or the ClusterQueue's cohort.
 
-                  Preemption can happen in two scenarios:
+                  Preemption may be configured to work in the following scenarios:
 
-                  - When a Workload fits within the nominal quota of the ClusterQueue, but
-                    the quota is currently borrowed by other ClusterQueues in the cohort.
-                    Preempting Workloads in other ClusterQueues allows this ClusterQueue to
-                    reclaim its nominal quota.
-                  - When a Workload doesn't fit within the nominal quota of the ClusterQueue
-                    and there are admitted Workloads in the ClusterQueue with lower priority.
+                    - When a Workload fits within the nominal quota of the ClusterQueue, but
+                      the quota is currently borrowed by other ClusterQueues in the cohort.
+                      We preempt workloads in other ClusterQueues to allow this ClusterQueue to
+                      reclaim its nominal quota. Configured using reclaimWithinCohort.
+                    - When a Workload doesn't fit within the nominal quota of the ClusterQueue
+                      and there are admitted Workloads in the ClusterQueue with lower priority.
+                      Configured using withinClusterQueue.
+                    - When a Workload may fit while both borrowing and preempting
+                      low priority workloads in the Cohort. Configured using borrowWithinCohort.
+                    - When FairSharing is enabled, to maintain fair distribution of
+                      unused resources. See FairSharing documentation.
 
                   The preemption algorithm tries to find a minimal set of Workloads to
-                  preempt to accommodate the pending Workload, preempting Workloads with
+                  preempt to accomomdate the pending Workload, preempting Workloads with
                   lower priority first.
                 properties:
                   borrowWithinCohort:
                     default: {}
                     description: |-
-                      borrowWithinCohort provides configuration to allow preemption within
-                      cohort while borrowing.
+                      BorrowWithinCohort contains configuration which allows to preempt workloads
+                      within cohort while borrowing. It only works with Classical Preemption,
+                      __not__ with Fair Sharing.
                     properties:
                       maxPriorityThreshold:
                         description: |-
diff --git a/config/components/webhook/manifests.yaml b/config/components/webhook/manifests.yaml
index c61cf7dc9..3c7b9ce6f 100644
--- a/config/components/webhook/manifests.yaml
+++ b/config/components/webhook/manifests.yaml
@@ -23,25 +23,6 @@ webhooks:
     resources:
     - appwrappers
   sideEffects: None
-- admissionReviewVersions:
-  - v1
-  clientConfig:
-    service:
-      name: webhook-service
-      namespace: system
-      path: /mutate--v1-pod
-  failurePolicy: Fail
-  name: mpod.kb.io
-  rules:
-  - apiGroups:
-    - ""
-    apiVersions:
-    - v1
-    operations:
-    - CREATE
-    resources:
-    - pods
-  sideEffects: None
 - admissionReviewVersions:
   - v1
   clientConfig:
@@ -176,6 +157,25 @@ webhooks:
     resources:
     - xgboostjobs
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /mutate--v1-pod
+  failurePolicy: Fail
+  name: mpod.kb.io
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - pods
+  sideEffects: None
 - admissionReviewVersions:
   - v1
   clientConfig:
@@ -356,26 +356,6 @@ webhooks:
     resources:
     - appwrappers
   sideEffects: None
-- admissionReviewVersions:
-  - v1
-  clientConfig:
-    service:
-      name: webhook-service
-      namespace: system
-      path: /validate--v1-pod
-  failurePolicy: Fail
-  name: vpod.kb.io
-  rules:
-  - apiGroups:
-    - ""
-    apiVersions:
-    - v1
-    operations:
-    - CREATE
-    - UPDATE
-    resources:
-    - pods
-  sideEffects: None
 - admissionReviewVersions:
   - v1
   clientConfig:
@@ -516,6 +496,26 @@ webhooks:
     resources:
     - xgboostjobs
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /validate--v1-pod
+  failurePolicy: Fail
+  name: vpod.kb.io
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    - UPDATE
+    resources:
+    - pods
+  sideEffects: None
 - admissionReviewVersions:
   - v1
   clientConfig:
diff --git a/keps/3899-remove-finalizers-with-strict-patch/README.md b/keps/3899-remove-finalizers-with-strict-patch/README.md
new file mode 100644
index 000000000..9d21d7905
--- /dev/null
+++ b/keps/3899-remove-finalizers-with-strict-patch/README.md
@@ -0,0 +1,89 @@
+# KEP-3899: Remove finalizers using strict patch
+
+<!-- toc -->
+- [Summary](#summary)
+- [Motivation](#motivation)
+  - [Goals](#goals)
+  - [Non-Goals](#non-goals)
+- [Proposal](#proposal)
+  - [User Stories (Optional)](#user-stories-optional)
+  - [Notes/Constraints/Caveats (Optional)](#notesconstraintscaveats-optional)
+  - [Risks and Mitigations](#risks-and-mitigations)
+- [Design Details](#design-details)
+  - [Test Plan](#test-plan)
+    - [Integration tests](#integration-tests)
+  - [Graduation Criteria](#graduation-criteria)
+- [Implementation History](#implementation-history)
+- [Drawbacks](#drawbacks)
+- [Alternatives](#alternatives)
+<!-- /toc -->
+
+## Summary
+
+Remove finalizers using strict patch.
+This would force every finalizer update to be handled individually, removing the possibility of
+race conditions.
+
+## Motivation
+
+Updates to the finalizers [could be lost if they were made at the time when
+Kueue removed its system finalizers](https://github.com/kubernetes-sigs/kueue/issues/3899). 
+This could result in the unexpected and faulty behavior of the resources. 
+To prevent this, we need to use strict mode that ensures the concurrently-safe finalizer handling.
+
+### Goals
+
+- Remove finalizers using a strict patch.
+
+
+### Non-Goals
+
+- Implement a feature to provide a knob if Kueue uses strict mode or not.
+
+## Proposal
+
+Make Kueue remove and update finalizers from resources using a strict patch to ensure
+concurrently-safe finalizer handling.
+
+
+### User Stories (Optional)
+
+
+### Notes/Constraints/Caveats (Optional)
+
+Removing finalizers using a strict patch will force the reconciler to requeue
+every change to the finalizers individually, which could result in a performance
+hit when used on a large scale.
+
+
+### Risks and Mitigations
+
+## Design Details
+
+### Test Plan
+
+[x] I/we understand the owners of the involved components may require updates to
+existing tests to make this code solid enough prior to committing the changes necessary
+to implement this enhancement.
+
+#### Integration tests
+
+- Test that verifies that no error happens when the strict mode is engaged.
+
+
+### Graduation Criteria
+
+GA/Stable:
+* Positive feedback from users
+* No critical performance hit
+* There are no significant performance disadvantages in manual performance testings by real K8s cluster or KWOK.
+
+## Implementation History
+
+KEP: 2025-02-26
+
+## Drawbacks
+
+
+## Alternatives
+
diff --git a/keps/3899-remove-finalizers-with-strict-patch/kep.yaml b/keps/3899-remove-finalizers-with-strict-patch/kep.yaml
new file mode 100644
index 000000000..2659dc8bd
--- /dev/null
+++ b/keps/3899-remove-finalizers-with-strict-patch/kep.yaml
@@ -0,0 +1,29 @@
+title: Remove finalizers using strict patch
+kep-number: 3899
+authors:
+  - "@mykysha"
+status: implementable
+creation-date: 2025-02-26
+reviewers:
+  - "@mimowo"
+  - "@tenzen-y"
+approvers:
+  - "@mimowo"
+  - "@tenzen-y"
+
+# The target maturity stage in the current dev cycle for this KEP.
+stage: beta
+
+# The most recent milestone for which work toward delivery of this KEP has been
+# done. This can be the current (upcoming) milestone, if it is being actively
+# worked on.
+latest-milestone: "v0.11"
+
+milestone:
+    beta: "v0.11"
+
+# The following PRR answers are required at alpha release
+# List the feature gate name and the components for which it must be enabled
+feature-gates:
+  - name: RemoveFinalizersWithStrictPatch
+disable-supported: true
diff --git a/pkg/cache/clusterqueue_snapshot.go b/pkg/cache/clusterqueue_snapshot.go
index cb419d7b1..54bb07a37 100644
--- a/pkg/cache/clusterqueue_snapshot.go
+++ b/pkg/cache/clusterqueue_snapshot.go
@@ -77,7 +77,7 @@ func (c *ClusterQueueSnapshot) SimulateUsageRemoval(workloads []*workload.Info)
 		usage = append(usage, w.Usage())
 	}
 	for _, u := range usage {
-		c.removeUsage(u)
+		c.RemoveUsage(u)
 	}
 	return func() {
 		for _, u := range usage {
@@ -93,7 +93,7 @@ func (c *ClusterQueueSnapshot) AddUsage(usage workload.Usage) {
 	c.updateTASUsage(usage.TAS, add)
 }
 
-func (c *ClusterQueueSnapshot) removeUsage(usage workload.Usage) {
+func (c *ClusterQueueSnapshot) RemoveUsage(usage workload.Usage) {
 	for fr, q := range usage.Quota {
 		removeUsage(c, fr, q)
 	}
diff --git a/pkg/cache/cohort_snapshot.go b/pkg/cache/cohort_snapshot.go
index f0c8c0795..b626483d3 100644
--- a/pkg/cache/cohort_snapshot.go
+++ b/pkg/cache/cohort_snapshot.go
@@ -67,6 +67,11 @@ func (c *CohortSnapshot) subtreeClusterQueueCount() int {
 	return count
 }
 
+func (c *CohortSnapshot) DominantResourceShare() int {
+	share, _ := dominantResourceShare(c, nil)
+	return share
+}
+
 // The methods below implement hierarchicalResourceNode interface.
 
 func (c *CohortSnapshot) getResourceNode() ResourceNode {
diff --git a/pkg/cache/snapshot.go b/pkg/cache/snapshot.go
index c19bd3a89..660376080 100644
--- a/pkg/cache/snapshot.go
+++ b/pkg/cache/snapshot.go
@@ -44,7 +44,7 @@ type Snapshot struct {
 func (s *Snapshot) RemoveWorkload(wl *workload.Info) {
 	cq := s.ClusterQueue(wl.ClusterQueue)
 	delete(cq.Workloads, workload.Key(wl.Obj))
-	cq.removeUsage(wl.Usage())
+	cq.RemoveUsage(wl.Usage())
 }
 
 // AddWorkload adds a workload from its corresponding ClusterQueue and
diff --git a/pkg/controller/jobs/deployment/deployment_webhook.go b/pkg/controller/jobs/deployment/deployment_webhook.go
index 8b4730f6e..8e58f586f 100644
--- a/pkg/controller/jobs/deployment/deployment_webhook.go
+++ b/pkg/controller/jobs/deployment/deployment_webhook.go
@@ -31,7 +31,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework/webhook"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/queue"
 )
 
@@ -77,7 +77,7 @@ func (wh *Webhook) Default(ctx context.Context, obj runtime.Object) error {
 		if deployment.Spec.Template.Annotations == nil {
 			deployment.Spec.Template.Annotations = make(map[string]string, 1)
 		}
-		deployment.Spec.Template.Annotations[pod.SuspendedByParentAnnotation] = FrameworkName
+		deployment.Spec.Template.Annotations[podconstants.SuspendedByParentAnnotation] = FrameworkName
 		if deployment.Spec.Template.Labels == nil {
 			deployment.Spec.Template.Labels = make(map[string]string, 1)
 		}
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
index 5bff53431..b171f96f1 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
@@ -35,7 +35,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	clientutil "sigs.k8s.io/kueue/pkg/util/client"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 )
@@ -72,13 +72,13 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 
 	if utilpod.IsTerminated(pod) {
 		err = client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, true, func() (bool, error) {
-			removed := controllerutil.RemoveFinalizer(pod, podcontroller.PodFinalizer)
+			removed := controllerutil.RemoveFinalizer(pod, podconstants.PodFinalizer)
 			if removed {
 				log.V(3).Info(
 					"Finalizing leaderworkerset pod in group",
 					"leaderworkerset", pod.Labels[leaderworkersetv1.SetNameLabelKey],
 					"pod", klog.KObj(pod),
-					"group", pod.Labels[podcontroller.GroupNameLabel],
+					"group", pod.Labels[podconstants.GroupNameLabel],
 				)
 			}
 			return removed, nil
@@ -90,7 +90,7 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 				return false, err
 			}
 			if updated {
-				log.V(3).Info("Updating pod in group", "pod", klog.KObj(pod), "group", pod.Labels[podcontroller.GroupNameLabel])
+				log.V(3).Info("Updating pod in group", "pod", klog.KObj(pod), "group", pod.Labels[podconstants.GroupNameLabel])
 			}
 			return updated, nil
 		}))
@@ -125,20 +125,20 @@ func (r *PodReconciler) setDefault(ctx context.Context, pod *corev1.Pod) (bool,
 	wlName := GetWorkloadName(lws.UID, lws.Name, pod.Labels[leaderworkersetv1.GroupIndexLabelKey])
 
 	pod.Labels[constants.QueueLabel] = queueName
-	pod.Labels[podcontroller.GroupNameLabel] = wlName
+	pod.Labels[podconstants.GroupNameLabel] = wlName
 	pod.Labels[constants.PrebuiltWorkloadLabel] = wlName
 	if priorityClass := jobframework.WorkloadPriorityClassName(lws); priorityClass != "" {
 		pod.Labels[constants.WorkloadPriorityClassLabel] = priorityClass
 	}
 
-	pod.Annotations[podcontroller.GroupTotalCountAnnotation] = fmt.Sprint(ptr.Deref(lws.Spec.LeaderWorkerTemplate.Size, 1))
-	pod.Annotations[podcontroller.RoleHashAnnotation] = string(kueue.DefaultPodSetName)
+	pod.Annotations[podconstants.GroupTotalCountAnnotation] = fmt.Sprint(ptr.Deref(lws.Spec.LeaderWorkerTemplate.Size, 1))
+	pod.Annotations[podconstants.RoleHashAnnotation] = string(kueue.DefaultPodSetName)
 
 	if lws.Spec.LeaderWorkerTemplate.LeaderTemplate != nil {
 		if _, ok := pod.Annotations[leaderworkersetv1.LeaderPodNameAnnotationKey]; !ok {
-			pod.Annotations[podcontroller.RoleHashAnnotation] = leaderPodSetName
+			pod.Annotations[podconstants.RoleHashAnnotation] = leaderPodSetName
 		} else {
-			pod.Annotations[podcontroller.RoleHashAnnotation] = workerPodSetName
+			pod.Annotations[podconstants.RoleHashAnnotation] = workerPodSetName
 		}
 	}
 
@@ -169,5 +169,5 @@ func (r *PodReconciler) handle(obj client.Object) bool {
 		return false
 	}
 	// Handle only leaderworkerset pods.
-	return pod.Annotations[podcontroller.SuspendedByParentAnnotation] == FrameworkName
+	return pod.Annotations[podconstants.SuspendedByParentAnnotation] == FrameworkName
 }
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
index ce21dcbaf..b1be89989 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
@@ -34,7 +34,7 @@ import (
 
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/queue"
 )
 
@@ -84,8 +84,8 @@ func (wh *Webhook) podTemplateSpecDefault(podTemplateSpec *corev1.PodTemplateSpe
 	if podTemplateSpec.Annotations == nil {
 		podTemplateSpec.Annotations = make(map[string]string, 1)
 	}
-	podTemplateSpec.Annotations[podcontroller.SuspendedByParentAnnotation] = FrameworkName
-	podTemplateSpec.Annotations[podcontroller.GroupServingAnnotation] = "true"
+	podTemplateSpec.Annotations[podconstants.SuspendedByParentAnnotation] = FrameworkName
+	podTemplateSpec.Annotations[podconstants.GroupServingAnnotationKey] = podconstants.GroupServingAnnotationValue
 }
 
 // +kubebuilder:webhook:path=/validate-leaderworkerset-x-k8s-io-v1-leaderworkerset,mutating=false,failurePolicy=fail,sideEffects=None,groups="leaderworkerset.x-k8s.io",resources=leaderworkersets,verbs=create;update,versions=v1,name=vleaderworkerset.kb.io,admissionReviewVersions=v1
diff --git a/pkg/controller/jobs/pod/constants/constants.go b/pkg/controller/jobs/pod/constants/constants.go
new file mode 100644
index 000000000..05ba6ab8c
--- /dev/null
+++ b/pkg/controller/jobs/pod/constants/constants.go
@@ -0,0 +1,38 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package constants
+
+import "sigs.k8s.io/kueue/pkg/constants"
+
+const (
+	PodFinalizer = constants.ManagedByKueueLabelKey
+
+	SchedulingGateName = "kueue.x-k8s.io/admission"
+
+	SuspendedByParentAnnotation       = "kueue.x-k8s.io/pod-suspending-parent"
+	GroupNameLabel                    = "kueue.x-k8s.io/pod-group-name"
+	GroupTotalCountAnnotation         = "kueue.x-k8s.io/pod-group-total-count"
+	GroupFastAdmissionAnnotationKey   = "kueue.x-k8s.io/pod-group-fast-admission"
+	GroupFastAdmissionAnnotationValue = "true"
+	GroupServingAnnotationKey         = "kueue.x-k8s.io/pod-group-serving"
+	GroupServingAnnotationValue       = "true"
+	RoleHashAnnotation                = "kueue.x-k8s.io/role-hash"
+	RetriableInGroupAnnotationKey     = "kueue.x-k8s.io/retriable-in-group"
+	RetriableInGroupAnnotationValue   = "false"
+	IsGroupWorkloadAnnotationKey      = "kueue.x-k8s.io/is-group-workload"
+	IsGroupWorkloadAnnotationValue    = "true"
+)
diff --git a/pkg/controller/jobs/pod/event_handlers.go b/pkg/controller/jobs/pod/event_handlers.go
index 57620b65d..8a47652a6 100644
--- a/pkg/controller/jobs/pod/event_handlers.go
+++ b/pkg/controller/jobs/pod/event_handlers.go
@@ -34,6 +34,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/util/expectations"
 )
 
@@ -42,7 +43,7 @@ var (
 )
 
 func reconcileRequestForPod(p *corev1.Pod) reconcile.Request {
-	groupName := p.GetLabels()[GroupNameLabel]
+	groupName := p.GetLabels()[podconstants.GroupNameLabel]
 
 	if groupName == "" {
 		return reconcile.Request{
@@ -82,7 +83,7 @@ func (h *podEventHandler) Delete(ctx context.Context, e event.DeleteEvent, q wor
 
 	log := ctrl.LoggerFrom(ctx).WithValues("pod", klog.KObj(p))
 
-	if g, isGroup := p.Labels[GroupNameLabel]; isGroup {
+	if g, isGroup := p.Labels[podconstants.GroupNameLabel]; isGroup {
 		// If the watch was temporarily unavailable, it is possible that the object reported in the event still
 		// has a finalizer, but we can consider this Pod cleaned up, as it is being deleted.
 		h.cleanedUpPodsExpectations.ObservedUID(log, types.NamespacedName{Namespace: p.Namespace, Name: g}, p.UID)
@@ -104,8 +105,8 @@ func (h *podEventHandler) queueReconcileForPod(ctx context.Context, object clien
 
 	log := ctrl.LoggerFrom(ctx).WithValues("pod", klog.KObj(p))
 
-	if g, isGroup := p.Labels[GroupNameLabel]; isGroup {
-		if !slices.Contains(p.Finalizers, PodFinalizer) {
+	if g, isGroup := p.Labels[podconstants.GroupNameLabel]; isGroup {
+		if !slices.Contains(p.Finalizers, podconstants.PodFinalizer) {
 			h.cleanedUpPodsExpectations.ObservedUID(log, types.NamespacedName{Namespace: p.Namespace, Name: g}, p.UID)
 		}
 	}
@@ -144,7 +145,7 @@ func (h *workloadHandler) queueReconcileForChildPod(ctx context.Context, object
 	log.V(5).Info("Queueing reconcile for parent pods")
 
 	// Compose request for a pod group if workload has an "is-group-workload" annotation
-	if w.Annotations[IsGroupWorkloadAnnotationKey] == IsGroupWorkloadAnnotationValue {
+	if w.Annotations[podconstants.IsGroupWorkloadAnnotationKey] == podconstants.IsGroupWorkloadAnnotationValue {
 		log.V(5).Info("Queueing reconcile for the pod group", "groupName", w.Name, "namespace", w.Namespace)
 		q.Add(reconcile.Request{
 			NamespacedName: types.NamespacedName{
diff --git a/pkg/controller/jobs/pod/indexer.go b/pkg/controller/jobs/pod/indexer.go
index 718d9b19a..804bc3ee9 100644
--- a/pkg/controller/jobs/pod/indexer.go
+++ b/pkg/controller/jobs/pod/indexer.go
@@ -19,6 +19,8 @@ package pod
 import (
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 )
 
 const (
@@ -31,7 +33,7 @@ func IndexPodGroupName(o client.Object) []string {
 		return nil
 	}
 
-	if labelValue, exists := pod.GetLabels()[GroupNameLabel]; exists {
+	if labelValue, exists := pod.GetLabels()[podconstants.GroupNameLabel]; exists {
 		return []string{labelValue}
 	}
 	return nil
diff --git a/pkg/controller/jobs/pod/pod_controller.go b/pkg/controller/jobs/pod/pod_controller.go
index d17699399..6cdc275f9 100644
--- a/pkg/controller/jobs/pod/pod_controller.go
+++ b/pkg/controller/jobs/pod/pod_controller.go
@@ -48,6 +48,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/podset"
 	clientutil "sigs.k8s.io/kueue/pkg/util/client"
 	"sigs.k8s.io/kueue/pkg/util/expectations"
@@ -58,12 +59,9 @@ import (
 )
 
 const (
-	SchedulingGateName             = "kueue.x-k8s.io/admission"
 	FrameworkName                  = "pod"
 	ConditionTypeTerminationTarget = "TerminationTarget"
 	errMsgIncorrectGroupRoleCount  = "pod group can't include more than 8 roles"
-	IsGroupWorkloadAnnotationKey   = "kueue.x-k8s.io/is-group-workload"
-	IsGroupWorkloadAnnotationValue = "true"
 )
 
 // Event reasons used by the pod controller
@@ -215,7 +213,7 @@ func podSuspended(p *corev1.Pod) bool {
 }
 
 func isUnretriablePod(pod corev1.Pod) bool {
-	return pod.Annotations[RetriableInGroupAnnotation] == "false"
+	return pod.Annotations[podconstants.RetriableInGroupAnnotationKey] == podconstants.RetriableInGroupAnnotationValue
 }
 
 // isUnretriableGroup returns true if at least one pod in the group
@@ -428,7 +426,7 @@ func (p *Pod) GVK() schema.GroupVersionKind {
 }
 
 func (p *Pod) PodLabelSelector() string {
-	return fmt.Sprintf("%s=%s", GroupNameLabel, p.pod.Labels[GroupNameLabel])
+	return fmt.Sprintf("%s=%s", podconstants.GroupNameLabel, p.pod.Labels[podconstants.GroupNameLabel])
 }
 
 func (p *Pod) Stop(ctx context.Context, c client.Client, _ []podset.PodSetInfo, stopReason jobframework.StopReason, eventMsg string) ([]client.Object, error) {
@@ -529,7 +527,7 @@ func (p *Pod) Finalize(ctx context.Context, c client.Client) error {
 	return parallelize.Until(ctx, len(podsInGroup.Items), func(i int) error {
 		pod := &podsInGroup.Items[i]
 		return clientutil.Patch(ctx, c, pod, false, func() (bool, error) {
-			return controllerutil.RemoveFinalizer(pod, PodFinalizer), nil
+			return controllerutil.RemoveFinalizer(pod, podconstants.PodFinalizer), nil
 		})
 	})
 }
@@ -545,20 +543,20 @@ func (p *Pod) Skip() bool {
 // podGroupName returns a value of GroupNameLabel for the pod object.
 // Returns an empty string if there's no such label.
 func podGroupName(p corev1.Pod) string {
-	return p.GetLabels()[GroupNameLabel]
+	return p.GetLabels()[podconstants.GroupNameLabel]
 }
 
 // groupTotalCount returns the value of GroupTotalCountAnnotation for the pod being reconciled at the moment.
 // It doesn't check if the whole group has the same total group count annotation value.
 func (p *Pod) groupTotalCount() (int, error) {
 	if podGroupName(p.pod) == "" {
-		return 0, fmt.Errorf("pod doesn't have a '%s' label", GroupNameLabel)
+		return 0, fmt.Errorf("pod doesn't have a '%s' label", podconstants.GroupNameLabel)
 	}
 
-	gtcAnnotation, ok := p.Object().GetAnnotations()[GroupTotalCountAnnotation]
+	gtcAnnotation, ok := p.Object().GetAnnotations()[podconstants.GroupTotalCountAnnotation]
 	if !ok {
 		return 0, fmt.Errorf("failed to extract '%s' annotation",
-			GroupTotalCountAnnotation)
+			podconstants.GroupTotalCountAnnotation)
 	}
 
 	gtc, err := strconv.Atoi(gtcAnnotation)
@@ -568,7 +566,7 @@ func (p *Pod) groupTotalCount() (int, error) {
 
 	if gtc < 1 {
 		return 0, fmt.Errorf("incorrect annotation value '%s=%s': group total count should be greater than zero",
-			GroupTotalCountAnnotation, gtcAnnotation)
+			podconstants.GroupTotalCountAnnotation, gtcAnnotation)
 	}
 
 	return gtc, nil
@@ -577,7 +575,7 @@ func (p *Pod) groupTotalCount() (int, error) {
 // getRoleHash will filter all the fields of the pod that are relevant to admission (pod role) and return a sha256
 // checksum of those fields. This is used to group the pods of the same roles when interacting with the workload.
 func getRoleHash(p corev1.Pod) (string, error) {
-	if roleHash, ok := p.Annotations[RoleHashAnnotation]; ok {
+	if roleHash, ok := p.Annotations[podconstants.RoleHashAnnotation]; ok {
 		return roleHash, nil
 	}
 	return utilpod.GenerateRoleHash(&p.Spec)
@@ -629,7 +627,7 @@ func (p *Pod) Load(ctx context.Context, c client.Client, key *types.NamespacedNa
 }
 
 func (p *Pod) constructGroupPodSets() ([]kueue.PodSet, error) {
-	if _, useFastAdmission := p.pod.GetAnnotations()[GroupFastAdmissionAnnotation]; useFastAdmission {
+	if _, useFastAdmission := p.pod.GetAnnotations()[podconstants.GroupFastAdmissionAnnotationKey]; useFastAdmission {
 		tc, err := p.groupTotalCount()
 		if err != nil {
 			return nil, err
@@ -724,7 +722,7 @@ func (p *Pod) validatePodGroupMetadata(r record.EventRecorder, activePods []core
 	}
 
 	originalQueue := jobframework.QueueName(p)
-	_, useFastAdmission := p.pod.GetAnnotations()[GroupFastAdmissionAnnotation]
+	_, useFastAdmission := p.pod.GetAnnotations()[podconstants.GroupFastAdmissionAnnotationKey]
 
 	if !useFastAdmission && len(activePods) < groupTotalCount {
 		errMsg := fmt.Sprintf("'%s' group has fewer runnable pods than expected", podGroupName(p.pod))
@@ -744,17 +742,17 @@ func (p *Pod) validatePodGroupMetadata(r record.EventRecorder, activePods []core
 				originalQueue, podInGroupQueue))
 		}
 
-		tc, err := strconv.Atoi(podInGroup.GetAnnotations()[GroupTotalCountAnnotation])
+		tc, err := strconv.Atoi(podInGroup.GetAnnotations()[podconstants.GroupTotalCountAnnotation])
 		if err != nil {
 			return fmt.Errorf("failed to extract '%s' annotation from the pod '%s': %w",
-				GroupTotalCountAnnotation,
+				podconstants.GroupTotalCountAnnotation,
 				podInGroup.GetName(),
 				err)
 		}
 		if tc != groupTotalCount {
 			return jobframework.UnretryableError(fmt.Sprintf("pods '%s' and '%s' has different '%s' values: %d!=%d",
 				p.pod.GetName(), podInGroup.GetName(),
-				GroupTotalCountAnnotation,
+				podconstants.GroupTotalCountAnnotation,
 				groupTotalCount, tc))
 		}
 	}
@@ -810,8 +808,8 @@ func sortInactivePods(clock clock.Clock, inactivePods []corev1.Pod) {
 	sort.Slice(inactivePods, func(i, j int) bool {
 		pi := &inactivePods[i]
 		pj := &inactivePods[j]
-		iFin := slices.Contains(pi.Finalizers, PodFinalizer)
-		jFin := slices.Contains(pj.Finalizers, PodFinalizer)
+		iFin := slices.Contains(pi.Finalizers, podconstants.PodFinalizer)
+		jFin := slices.Contains(pj.Finalizers, podconstants.PodFinalizer)
 		if iFin != jFin {
 			return iFin
 		}
@@ -835,8 +833,8 @@ func sortActivePods(activePods []corev1.Pod) {
 	sort.Slice(activePods, func(i, j int) bool {
 		pi := &activePods[i]
 		pj := &activePods[j]
-		iFin := slices.Contains(pi.Finalizers, PodFinalizer)
-		jFin := slices.Contains(pj.Finalizers, PodFinalizer)
+		iFin := slices.Contains(pi.Finalizers, podconstants.PodFinalizer)
+		jFin := slices.Contains(pj.Finalizers, podconstants.PodFinalizer)
 		// Prefer to keep pods that have a finalizer.
 		if iFin != jFin {
 			return iFin
@@ -866,7 +864,7 @@ func (p *Pod) removeExcessPods(ctx context.Context, c client.Client, r record.Ev
 	err := parallelize.Until(ctx, len(extraPods), func(i int) error {
 		pod := extraPods[i]
 		if err := clientutil.Patch(ctx, c, &pod, false, func() (bool, error) {
-			removed := controllerutil.RemoveFinalizer(&pod, PodFinalizer)
+			removed := controllerutil.RemoveFinalizer(&pod, podconstants.PodFinalizer)
 			log.V(3).Info("Finalizing excess pod in group", "excessPod", klog.KObj(&pod))
 			return removed, nil
 		}); err != nil {
@@ -906,7 +904,7 @@ func (p *Pod) finalizePods(ctx context.Context, c client.Client, extraPods []cor
 		pod := extraPods[i]
 		var removed bool
 		if err := clientutil.Patch(ctx, c, &pod, false, func() (bool, error) {
-			removed = controllerutil.RemoveFinalizer(&pod, PodFinalizer)
+			removed = controllerutil.RemoveFinalizer(&pod, podconstants.PodFinalizer)
 			log.V(3).Info("Finalizing pod in group", "Pod", klog.KObj(&pod))
 			return removed, nil
 		}); err != nil {
@@ -1207,7 +1205,7 @@ func (p *Pod) equivalentToWorkload(wl *kueue.Workload, jobPodSets []kueue.PodSet
 }
 
 func (p *Pod) isServing() bool {
-	return p.isGroup && p.pod.Annotations[GroupServingAnnotation] == "true"
+	return p.isGroup && p.pod.Annotations[podconstants.GroupServingAnnotationKey] == podconstants.GroupServingAnnotationValue
 }
 
 func (p *Pod) isReclaimable() bool {
@@ -1335,19 +1333,19 @@ func GetWorkloadNameForPod(podName string, podUID types.UID) string {
 
 func NewGroupWorkload(name string, obj client.Object, podSets []kueue.PodSet, labelKeysToCopy []string) *kueue.Workload {
 	wl := jobframework.NewWorkload(name, obj, podSets, labelKeysToCopy)
-	wl.Annotations[IsGroupWorkloadAnnotationKey] = IsGroupWorkloadAnnotationValue
+	wl.Annotations[podconstants.IsGroupWorkloadAnnotationKey] = podconstants.IsGroupWorkloadAnnotationValue
 
 	return wl
 }
 
 func isGated(pod *corev1.Pod) bool {
-	return utilpod.HasGate(pod, SchedulingGateName)
+	return utilpod.HasGate(pod, podconstants.SchedulingGateName)
 }
 
 func ungate(pod *corev1.Pod) bool {
-	return utilpod.Ungate(pod, SchedulingGateName)
+	return utilpod.Ungate(pod, podconstants.SchedulingGateName)
 }
 
 func gate(pod *corev1.Pod) bool {
-	return utilpod.Gate(pod, SchedulingGateName)
+	return utilpod.Gate(pod, podconstants.SchedulingGateName)
 }
diff --git a/pkg/controller/jobs/pod/pod_webhook.go b/pkg/controller/jobs/pod/pod_webhook.go
index 31ffbe587..9cac5045e 100644
--- a/pkg/controller/jobs/pod/pod_webhook.go
+++ b/pkg/controller/jobs/pod/pod_webhook.go
@@ -40,31 +40,21 @@ import (
 	ctrlconstants "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework/webhook"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 )
 
-const (
-	PodFinalizer                 = constants.ManagedByKueueLabelKey
-	GroupNameLabel               = "kueue.x-k8s.io/pod-group-name"
-	GroupTotalCountAnnotation    = "kueue.x-k8s.io/pod-group-total-count"
-	GroupFastAdmissionAnnotation = "kueue.x-k8s.io/pod-group-fast-admission"
-	GroupServingAnnotation       = "kueue.x-k8s.io/pod-group-serving"
-	SuspendedByParentAnnotation  = "kueue.x-k8s.io/pod-suspending-parent"
-	RoleHashAnnotation           = "kueue.x-k8s.io/role-hash"
-	RetriableInGroupAnnotation   = "kueue.x-k8s.io/retriable-in-group"
-)
-
 var (
 	metaPath                       = field.NewPath("metadata")
 	labelsPath                     = metaPath.Child("labels")
 	annotationsPath                = metaPath.Child("annotations")
 	managedLabelPath               = labelsPath.Key(constants.ManagedByKueueLabelKey)
-	groupNameLabelPath             = labelsPath.Key(GroupNameLabel)
+	groupNameLabelPath             = labelsPath.Key(podconstants.GroupNameLabel)
 	prebuiltWorkloadLabelPath      = labelsPath.Key(ctrlconstants.PrebuiltWorkloadLabel)
-	groupTotalCountAnnotationPath  = annotationsPath.Key(GroupTotalCountAnnotation)
-	retriableInGroupAnnotationPath = annotationsPath.Key(RetriableInGroupAnnotation)
+	groupTotalCountAnnotationPath  = annotationsPath.Key(podconstants.GroupTotalCountAnnotation)
+	retriableInGroupAnnotationPath = annotationsPath.Key(podconstants.RetriableInGroupAnnotationKey)
 
 	errPodOptsTypeAssertion = errors.New("options are not of type PodIntegrationOptions")
 )
@@ -131,7 +121,7 @@ func (p *Pod) addRoleHash() error {
 		return err
 	}
 
-	p.pod.Annotations[RoleHashAnnotation] = hash
+	p.pod.Annotations[podconstants.RoleHashAnnotation] = hash
 	return nil
 }
 
@@ -140,7 +130,7 @@ func (w *PodWebhook) Default(ctx context.Context, obj runtime.Object) error {
 	log := ctrl.LoggerFrom(ctx).WithName("pod-webhook")
 	log.V(5).Info("Applying defaults")
 
-	_, suspend := pod.pod.GetAnnotations()[SuspendedByParentAnnotation]
+	_, suspend := pod.pod.GetAnnotations()[podconstants.SuspendedByParentAnnotation]
 	if !suspend {
 		// Namespace filtering
 		ns := corev1.Namespace{}
@@ -204,7 +194,7 @@ func (w *PodWebhook) Default(ctx context.Context, obj runtime.Object) error {
 	}
 
 	if suspend {
-		controllerutil.AddFinalizer(pod.Object(), PodFinalizer)
+		controllerutil.AddFinalizer(pod.Object(), podconstants.PodFinalizer)
 
 		if pod.pod.Labels == nil {
 			pod.pod.Labels = make(map[string]string)
@@ -273,7 +263,7 @@ func (w *PodWebhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.
 		allErrs = append(allErrs, validation.ValidateImmutableField(podGroupName(newPod.pod), podGroupName(oldPod.pod), groupNameLabelPath)...)
 	}
 
-	if _, suspendByParent := newPod.pod.Annotations[SuspendedByParentAnnotation]; !suspendByParent {
+	if _, suspendByParent := newPod.pod.Annotations[podconstants.SuspendedByParentAnnotation]; !suspendByParent {
 		if warn := warningForPodManagedLabel(newPod); warn != "" {
 			warnings = append(warnings, warn)
 		}
@@ -318,22 +308,22 @@ func warningForPodManagedLabel(p *Pod) string {
 func validatePodGroupMetadata(p *Pod) field.ErrorList {
 	var allErrs field.ErrorList
 
-	gtc, gtcExists := p.pod.GetAnnotations()[GroupTotalCountAnnotation]
+	gtc, gtcExists := p.pod.GetAnnotations()[podconstants.GroupTotalCountAnnotation]
 
 	if podGroupName(p.pod) == "" {
 		if gtcExists {
 			return append(allErrs, field.Required(
 				groupNameLabelPath,
-				fmt.Sprintf("both the '%s' annotation and the '%s' label should be set", GroupTotalCountAnnotation, GroupNameLabel),
+				fmt.Sprintf("both the '%s' annotation and the '%s' label should be set", podconstants.GroupTotalCountAnnotation, podconstants.GroupNameLabel),
 			))
 		}
 	} else {
-		allErrs = append(allErrs, jobframework.ValidateLabelAsCRDName(p.Object(), GroupNameLabel)...)
+		allErrs = append(allErrs, jobframework.ValidateLabelAsCRDName(p.Object(), podconstants.GroupNameLabel)...)
 
 		if !gtcExists {
 			return append(allErrs, field.Required(
 				groupTotalCountAnnotationPath,
-				fmt.Sprintf("both the '%s' annotation and the '%s' label should be set", GroupTotalCountAnnotation, GroupNameLabel),
+				fmt.Sprintf("both the '%s' annotation and the '%s' label should be set", podconstants.GroupTotalCountAnnotation, podconstants.GroupNameLabel),
 			))
 		}
 	}
diff --git a/pkg/controller/jobs/statefulset/statefulset_reconciler.go b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
index bee918085..c8b2464c7 100644
--- a/pkg/controller/jobs/statefulset/statefulset_reconciler.go
+++ b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
@@ -36,7 +36,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	clientutil "sigs.k8s.io/kueue/pkg/util/client"
 	"sigs.k8s.io/kueue/pkg/util/parallelize"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
diff --git a/pkg/controller/jobs/statefulset/statefulset_webhook.go b/pkg/controller/jobs/statefulset/statefulset_webhook.go
index f1f0dc1f0..e48939645 100644
--- a/pkg/controller/jobs/statefulset/statefulset_webhook.go
+++ b/pkg/controller/jobs/statefulset/statefulset_webhook.go
@@ -34,7 +34,7 @@ import (
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/queue"
 )
 
@@ -78,17 +78,17 @@ func (wh *Webhook) Default(ctx context.Context, obj runtime.Object) error {
 		if ss.Spec.Template.Annotations == nil {
 			ss.Spec.Template.Annotations = make(map[string]string, 1)
 		}
-		ss.Spec.Template.Annotations[pod.SuspendedByParentAnnotation] = FrameworkName
+		ss.Spec.Template.Annotations[podconstants.SuspendedByParentAnnotation] = FrameworkName
 		queueName := jobframework.QueueNameForObject(ss.Object())
 		if queueName != "" {
 			if ss.Spec.Template.Labels == nil {
 				ss.Spec.Template.Labels = make(map[string]string, 2)
 			}
 			ss.Spec.Template.Labels[constants.QueueLabel] = queueName
-			ss.Spec.Template.Labels[pod.GroupNameLabel] = GetWorkloadName(ss.Name)
-			ss.Spec.Template.Annotations[pod.GroupTotalCountAnnotation] = fmt.Sprint(ptr.Deref(ss.Spec.Replicas, 1))
-			ss.Spec.Template.Annotations[pod.GroupFastAdmissionAnnotation] = "true"
-			ss.Spec.Template.Annotations[pod.GroupServingAnnotation] = "true"
+			ss.Spec.Template.Labels[podconstants.GroupNameLabel] = GetWorkloadName(ss.Name)
+			ss.Spec.Template.Annotations[podconstants.GroupTotalCountAnnotation] = fmt.Sprint(ptr.Deref(ss.Spec.Replicas, 1))
+			ss.Spec.Template.Annotations[podconstants.GroupFastAdmissionAnnotationKey] = podconstants.GroupFastAdmissionAnnotationValue
+			ss.Spec.Template.Annotations[podconstants.GroupServingAnnotationKey] = podconstants.GroupServingAnnotationValue
 			ss.Spec.Template.Annotations[kueuealpha.PodGroupPodIndexLabelAnnotation] = appsv1.PodIndexLabel
 		}
 		if priorityClass := jobframework.WorkloadPriorityClassName(ss.Object()); priorityClass != "" {
diff --git a/pkg/hierarchy/cohort.go b/pkg/hierarchy/cohort.go
index e6f2ca79f..72d66a507 100644
--- a/pkg/hierarchy/cohort.go
+++ b/pkg/hierarchy/cohort.go
@@ -45,6 +45,11 @@ func (c *Cohort[CQ, C]) ChildCohorts() []C {
 	return c.childCohorts.UnsortedList()
 }
 
+// ChildCount returns number of Cohorts + ClusterQueues.
+func (c *Cohort[CQ, C]) ChildCount() int {
+	return c.childCohorts.Len() + c.childCqs.Len()
+}
+
 func NewCohort[CQ, C nodeBase]() Cohort[CQ, C] {
 	return Cohort[CQ, C]{
 		childCohorts: sets.New[C](),
diff --git a/pkg/scheduler/fair_sharing_iterator.go b/pkg/scheduler/fair_sharing_iterator.go
new file mode 100644
index 000000000..0ed35f2f1
--- /dev/null
+++ b/pkg/scheduler/fair_sharing_iterator.go
@@ -0,0 +1,230 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package scheduler
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/go-logr/logr"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/features"
+	"sigs.k8s.io/kueue/pkg/util/priority"
+	"sigs.k8s.io/kueue/pkg/workload"
+)
+
+// fairSharingIterator orders candidates in a "fair" manner for
+// consideration by scheduling when FairSharing is enabled. See
+// runTournament for description of algorithm.
+type fairSharingIterator struct {
+	// cqToEntry tracks ClusterQueues which still have workloads
+	// to schedule, and the corresponding workload entry.
+	cqToEntry     map[*cache.ClusterQueueSnapshot]*entry
+	entryComparer entryComparer
+	log           logr.Logger
+}
+
+func makeFairSharingIterator(ctx context.Context, entries []entry, workloadOrdering workload.Ordering) *fairSharingIterator {
+	f := fairSharingIterator{
+		cqToEntry: make(map[*cache.ClusterQueueSnapshot]*entry, len(entries)),
+		entryComparer: entryComparer{
+			workloadOrdering: workloadOrdering,
+		},
+		log: ctrl.LoggerFrom(ctx),
+	}
+	for i := range entries {
+		f.cqToEntry[entries[i].clusterQueueSnapshot] = &entries[i]
+	}
+	return &f
+}
+
+func (f *fairSharingIterator) hasNext() bool {
+	return len(f.cqToEntry) > 0
+}
+
+func (f *fairSharingIterator) pop() *entry {
+	cq := f.getCq()
+
+	// CQ has no Cohort. We simply return its workload.
+	if !cq.HasParent() {
+		entry := f.cqToEntry[cq]
+		f.log.V(3).Info("Returning workload from ClusterQueue without Cohort",
+			"clusterQueue", klog.KRef("", cq.GetName()),
+			"workload", klog.KObj(entry.Obj))
+		delete(f.cqToEntry, cq)
+		return entry
+	}
+
+	// CQ is part of a Cohort. We run a tournament, to select the
+	// most fair workload at each level.
+	root := cq.Parent().Root()
+	log := f.log.WithValues("rootCohort", klog.KRef("", root.GetName()))
+
+	log.V(5).Info("Computing DominantResourceShare for tournament")
+	f.entryComparer.computeDRS(root, f.cqToEntry)
+
+	log.V(3).Info("Running tournament to decide next workload to consider in scheduling cycle")
+	entry := runTournament(root, f.entryComparer, f.cqToEntry)
+
+	log = log.WithValues(
+		"cohort", klog.KRef("", entry.clusterQueueSnapshot.Parent().GetName()),
+		"clusterQueue", klog.KRef("", entry.clusterQueueSnapshot.GetName()),
+		"winningWorkload", klog.KObj(entry.Obj))
+
+	log.V(3).Info("Determined tournament winner")
+	f.entryComparer.logDrsValuesWhenVerbose(log)
+
+	delete(f.cqToEntry, entry.clusterQueueSnapshot)
+	return entry
+}
+
+// getCq returns a CQ with a workload pending scheduling. This
+// function is non-deterministic. We don't have any guarantees on
+// scheduling order of workloads in different Cohort. Workload
+// consideration is nearly deterministic within Cohort (only when DRS,
+// Priority, and time are equal it is non-deterministic).
+func (f *fairSharingIterator) getCq() *cache.ClusterQueueSnapshot {
+	for cq := range f.cqToEntry {
+		return cq
+	}
+	return nil
+}
+
+// runTournament is a recursive algorithm which nominates one workload
+// for each Cohort.  It compares the DominantResourceShare (DRS) value
+// of each of the Cohort's children nodes (CQs or Cohorts), including
+// in this DRS value the workload which that child node is
+// nominating. The node with the lowest DRS wins, with some additional
+// tiebreaks (see entryComparer.less).
+//
+// This process results in one workload (or zero if Cohort has no
+// remaining workloads to schedule this cycle) being bubbled up per
+// node, until exactly one workload remains at the root.
+func runTournament(cohort *cache.CohortSnapshot, ec entryComparer, cqToEntry map[*cache.ClusterQueueSnapshot]*entry) *entry {
+	candidates := make([]*entry, 0, cohort.ChildCount())
+
+	// Run algorithm recursively for each of the child Cohorts.
+	for _, childCohort := range cohort.ChildCohorts() {
+		// The tournament returns 0 nodes, when the child
+		// Cohort has no workloads left to be scheduled this
+		// cycle.
+		if candidate := runTournament(childCohort, ec, cqToEntry); candidate != nil {
+			candidates = append(candidates, candidate)
+		}
+	}
+
+	// Collect entries from CQ. If an entry was returned during a
+	// previous call to pop, it will not be in the cqToEntry map.
+	for _, childCq := range cohort.ChildCQs() {
+		if candidate, ok := cqToEntry[childCq]; ok {
+			candidates = append(candidates, candidate)
+		}
+	}
+
+	if len(candidates) == 0 {
+		return nil
+	}
+
+	// Compare DRS values for each workload, for each of the
+	// children of the current Cohort.
+	best := candidates[0]
+	for _, current := range candidates[1:] {
+		if ec.less(current, best, cohort.GetName()) {
+			best = current
+		}
+	}
+	return best
+}
+
+type drsKey struct {
+	parentCohort string
+	workloadKey  string
+}
+
+type entryComparer struct {
+	drsValues        map[drsKey]int
+	workloadOrdering workload.Ordering
+}
+
+func (e *entryComparer) less(a, b *entry, parentCohort string) bool {
+	aDrs := e.drsValues[drsKey{parentCohort: parentCohort, workloadKey: workload.Key(a.Obj)}]
+	bDrs := e.drsValues[drsKey{parentCohort: parentCohort, workloadKey: workload.Key(b.Obj)}]
+	// 1: DRF
+	if aDrs != bDrs {
+		return aDrs < bDrs
+	}
+
+	// 2: Priority
+	if features.Enabled(features.PrioritySortingWithinCohort) {
+		p1 := priority.Priority(a.Obj)
+		p2 := priority.Priority(b.Obj)
+		if p1 != p2 {
+			return p1 > p2
+		}
+	}
+
+	// 3: FIFO
+	aComparisonTimestamp := e.workloadOrdering.GetQueueOrderTimestamp(a.Obj)
+	bComparisonTimestamp := e.workloadOrdering.GetQueueOrderTimestamp(b.Obj)
+	return aComparisonTimestamp.Before(bComparisonTimestamp)
+}
+
+// computeDRS calculates DominantResourceShare (DRS) for each node
+// after admission of workload, for all nodes on path from CQ to
+// root-1.  During the tournament, these values are used to compare
+// all children the parentCohort, to select the child with the lowest
+// DRS after admission of its nominated workload.
+func (ec *entryComparer) computeDRS(rootCohort *cache.CohortSnapshot, cqToEntry map[*cache.ClusterQueueSnapshot]*entry) {
+	ec.drsValues = make(map[drsKey]int)
+	for _, cq := range rootCohort.SubtreeClusterQueues() {
+		entry, ok := cqToEntry[cq]
+		if !ok {
+			continue
+		}
+		// We add workload's usage to CQ, so that all
+		// subsequent DRS include the admission of workload.
+		cq.AddUsage(entry.assignmentUsage())
+
+		// calculate DRS, with workload, for CQ.
+		dominantResourceShare := cq.DominantResourceShare()
+		ec.drsValues[drsKey{parentCohort: cq.Parent().GetName(), workloadKey: workload.Key(entry.Obj)}] = dominantResourceShare
+
+		// calculate DRS, with workload, for all Cohorts on
+		// path to root.
+		cohort := cq.Parent()
+		for cohort.HasParent() {
+			dominantResourceShare := cohort.DominantResourceShare()
+			ec.drsValues[drsKey{parentCohort: cohort.Parent().GetName(), workloadKey: workload.Key(entry.Obj)}] = dominantResourceShare
+			cohort = cohort.Parent()
+		}
+
+		cq.RemoveUsage(entry.assignmentUsage())
+	}
+}
+
+func (ec *entryComparer) logDrsValuesWhenVerbose(log logr.Logger) {
+	if logV := log.V(5); logV.Enabled() {
+		serializableDrs := make([]string, 0, len(ec.drsValues))
+		for k, v := range ec.drsValues {
+			serializableDrs = append(serializableDrs, fmt.Sprintf("{parentCohort: %s, workload %s, drs: %d}", k.parentCohort, k.workloadKey, v))
+		}
+		logV.Info("DominantResourceShare values used during tournament", "drsValues", serializableDrs)
+	}
+}
diff --git a/pkg/scheduler/flavorassigner/flavorassigner.go b/pkg/scheduler/flavorassigner/flavorassigner.go
index dae18168a..6ab1cc114 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner.go
@@ -47,7 +47,7 @@ type Assignment struct {
 
 	// Usage is the accumulated Usage of resources as pod sets get
 	// flavors assigned.
-	Usage resources.FlavorResourceQuantities
+	Usage workload.Usage
 
 	// representativeMode is the cached representative mode for this assignment.
 	representativeMode *FlavorAssignmentMode
@@ -59,10 +59,11 @@ func (assignment *Assignment) UpdateForTASResult(result cache.TASAssignmentsResu
 		psAssignment := assignment.podSetAssignmentByName(psName)
 		psAssignment.TopologyAssignment = psResult.TopologyAssignment
 	}
+	assignment.Usage.TAS = assignment.computeTASUsage()
 }
 
 // TASUsage computes the TAS usage for the assignment
-func (assignment *Assignment) TASUsage() workload.TASUsage {
+func (assignment *Assignment) computeTASUsage() workload.TASUsage {
 	result := make(workload.TASUsage)
 	for _, psa := range assignment.PodSets {
 		if psa.TopologyAssignment != nil {
@@ -389,7 +390,9 @@ func (a *FlavorAssigner) assignFlavors(log logr.Logger, counts []int32) Assignme
 	}
 	assignment := Assignment{
 		PodSets: make([]PodSetAssignment, 0, len(requests)),
-		Usage:   make(resources.FlavorResourceQuantities),
+		Usage: workload.Usage{
+			Quota: make(resources.FlavorResourceQuantities),
+		},
 		LastState: workload.AssignmentClusterQueueState{
 			LastTriedFlavorIdx:     make([]map[corev1.ResourceName]int, 0, len(requests)),
 			ClusterQueueGeneration: a.cq.AllocatableResourceGeneration,
@@ -414,7 +417,7 @@ func (a *FlavorAssigner) assignFlavors(log logr.Logger, counts []int32) Assignme
 				// No need to compute again.
 				continue
 			}
-			flavors, status := a.findFlavorForPodSetResource(log, i, podSet.Requests, resName, assignment.Usage)
+			flavors, status := a.findFlavorForPodSetResource(log, i, podSet.Requests, resName, assignment.Usage.Quota)
 			if status.IsError() || len(flavors) == 0 {
 				psAssignment.Flavors = nil
 				psAssignment.Status = status
@@ -482,7 +485,7 @@ func (a *Assignment) append(requests resources.Requests, psAssignment *PodSetAss
 			a.Borrowing = true
 		}
 		fr := resources.FlavorResource{Flavor: flvAssignment.Name, Resource: resource}
-		a.Usage[fr] += requests[resource]
+		a.Usage.Quota[fr] += requests[resource]
 		flavorIdx[resource] = flvAssignment.TriedFlavorIdx
 	}
 	a.LastState.LastTriedFlavorIdx = append(a.LastState.LastTriedFlavorIdx, flavorIdx)
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 948e88a9e..1d0718cec 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -157,16 +157,15 @@ func (p *Preemptor) getTargets(preemptionCtx *preemptionCtx) []*Target {
 		return minimalPreemptions(preemptionCtx, candidates, true, nil)
 	}
 
-	borrowWithinCohort, thresholdPrio := canBorrowWithinCohort(preemptionCtx)
 	if p.enableFairSharing {
-		return p.fairPreemptions(preemptionCtx, candidates, thresholdPrio)
+		return p.fairPreemptions(preemptionCtx, candidates)
 	}
 	// There is a potential of preemption of workloads from the other queue in the
 	// cohort. We proceed with borrowing only if the dedicated policy
 	// (borrowWithinCohort) is enabled. This ensures the preempted workloads
 	// have lower priority, and so they will not preempt the preemptor when
 	// requeued.
-	if borrowWithinCohort {
+	if borrowWithinCohort, thresholdPrio := canBorrowWithinCohort(preemptionCtx); borrowWithinCohort {
 		if !queueUnderNominalInResourcesNeedingPreemption(preemptionCtx) {
 			// It can only preempt workloads from another CQ if they are strictly under allowBorrowingBelowPriority.
 			candidates = candidatesFromCQOrUnderThreshold(candidates, preemptionCtx.preemptor.ClusterQueue, *thresholdPrio)
@@ -357,9 +356,9 @@ func parseStrategies(s []config.PreemptionStrategy) []fsStrategy {
 	return strategies
 }
 
-func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info, allowBorrowingBelowPriority *int32) []*Target {
+func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info) []*Target {
 	if logV := preemptionCtx.log.V(5); logV.Enabled() {
-		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "allowBorrowingBelowPriority", allowBorrowingBelowPriority, "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
+		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
 	}
 	requests := preemptionCtx.requests
 	cqHeap := cqHeapFromCandidates(candidates, false, preemptionCtx.snapshot)
@@ -391,15 +390,11 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*
 		}
 
 		for i, candWl := range candCQ.workloads {
-			belowThreshold := allowBorrowingBelowPriority != nil && priority.Priority(candWl.Obj) < *allowBorrowingBelowPriority
 			newCandShareVal := candCQ.cq.DominantResourceShareWithout(candWl.FlavorResourceUsage())
 			strategy := p.fsStrategies[0](newNominatedShareValue, candCQ.share, newCandShareVal)
-			if belowThreshold || strategy {
+			if strategy {
 				preemptionCtx.snapshot.RemoveWorkload(candWl)
 				reason := kueue.InCohortFairSharingReason
-				if !strategy {
-					reason = kueue.InCohortReclaimWhileBorrowingReason
-				}
 
 				targets = append(targets, &Target{
 					WorkloadInfo: candWl,
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 9abab7437..4c93d4e6c 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -199,7 +199,7 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 	entries := s.nominate(ctx, headWorkloads, snapshot)
 
 	// 4. Create iterator which returns ordered entries.
-	var iterator entryIterator = makeIterator(entries, s.workloadOrdering, s.fairSharing.Enable)
+	iterator := makeIterator(ctx, entries, s.workloadOrdering, s.fairSharing.Enable)
 
 	// 5. Admit entries, ensuring that no more than one workload gets
 	// admitted by a cohort (if borrowing).
@@ -210,15 +210,21 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 	skippedPreemptions := make(map[string]int)
 	for iterator.hasNext() {
 		e := iterator.pop()
-		mode := e.assignment.RepresentativeMode()
-		if mode == flavorassigner.NoFit {
-			continue
-		}
 
 		cq := snapshot.ClusterQueue(e.ClusterQueue)
 		log := log.WithValues("workload", klog.KObj(e.Obj), "clusterQueue", klog.KRef("", e.ClusterQueue))
+		if cq.HasParent() {
+			log = log.WithValues("parentCohort", klog.KRef("", cq.Parent().GetName()), "rootCohort", klog.KRef("", cq.Parent().Root().GetName()))
+		}
 		ctx := ctrl.LoggerInto(ctx, log)
 
+		mode := e.assignment.RepresentativeMode()
+		if mode == flavorassigner.NoFit {
+			log.V(3).Info("Skipping workload as FlavorAssigner assigned NoFit mode")
+			continue
+		}
+		log.V(2).Info("Attempting to schedule workload")
+
 		if mode == flavorassigner.Preempt && len(e.preemptionTargets) == 0 {
 			log.V(2).Info("Workload requires preemption, but there are no candidate workloads allowed for preemption", "preemption", cq.Preemption)
 			// we use resourcesToReserve to block capacity up to either the nominal capacity,
@@ -235,7 +241,7 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 			continue
 		}
 
-		usage := e.usage()
+		usage := e.assignmentUsage()
 		if !fits(cq, &usage, preemptedWorkloads, e.preemptionTargets) {
 			setSkipped(e, "Workload no longer fits after processing another workload")
 			if mode == flavorassigner.Preempt {
@@ -313,19 +319,16 @@ type entry struct {
 	// workload.Info holds the workload from the API as well as resource usage
 	// and flavors assigned.
 	workload.Info
-	dominantResourceShare int
-	assignment            flavorassigner.Assignment
-	status                entryStatus
-	inadmissibleMsg       string
-	requeueReason         queue.RequeueReason
-	preemptionTargets     []*preemption.Target
+	assignment           flavorassigner.Assignment
+	status               entryStatus
+	inadmissibleMsg      string
+	requeueReason        queue.RequeueReason
+	preemptionTargets    []*preemption.Target
+	clusterQueueSnapshot *cache.ClusterQueueSnapshot
 }
 
-func (e *entry) usage() workload.Usage {
-	return workload.Usage{
-		Quota: e.assignment.Usage,
-		TAS:   e.assignment.TASUsage(),
-	}
+func (e *entry) assignmentUsage() workload.Usage {
+	return e.assignment.Usage
 }
 
 // nominate returns the workloads with their requirements (resource flavors, borrowing) if
@@ -335,9 +338,9 @@ func (s *Scheduler) nominate(ctx context.Context, workloads []workload.Info, sna
 	entries := make([]entry, 0, len(workloads))
 	for _, w := range workloads {
 		log := log.WithValues("workload", klog.KObj(w.Obj), "clusterQueue", klog.KRef("", w.ClusterQueue))
-		cq := snap.ClusterQueue(w.ClusterQueue)
 		ns := corev1.Namespace{}
 		e := entry{Info: w}
+		e.clusterQueueSnapshot = snap.ClusterQueue(w.ClusterQueue)
 		if s.cache.IsAssumedOrAdmittedWorkload(w) {
 			log.Info("Workload skipped from admission because it's already assumed or admitted", "workload", klog.KObj(w.Obj))
 			continue
@@ -345,11 +348,11 @@ func (s *Scheduler) nominate(ctx context.Context, workloads []workload.Info, sna
 			e.inadmissibleMsg = "The workload has failed admission checks"
 		} else if snap.InactiveClusterQueueSets.Has(w.ClusterQueue) {
 			e.inadmissibleMsg = fmt.Sprintf("ClusterQueue %s is inactive", w.ClusterQueue)
-		} else if cq == nil {
+		} else if e.clusterQueueSnapshot == nil {
 			e.inadmissibleMsg = fmt.Sprintf("ClusterQueue %s not found", w.ClusterQueue)
 		} else if err := s.client.Get(ctx, types.NamespacedName{Name: w.Obj.Namespace}, &ns); err != nil {
 			e.inadmissibleMsg = fmt.Sprintf("Could not obtain workload namespace: %v", err)
-		} else if !cq.NamespaceSelector.Matches(labels.Set(ns.Labels)) {
+		} else if !e.clusterQueueSnapshot.NamespaceSelector.Matches(labels.Set(ns.Labels)) {
 			e.inadmissibleMsg = "Workload namespace doesn't match ClusterQueue selector"
 			e.requeueReason = queue.RequeueReasonNamespaceMismatch
 		} else if err := workload.ValidateResources(&w); err != nil {
@@ -360,9 +363,6 @@ func (s *Scheduler) nominate(ctx context.Context, workloads []workload.Info, sna
 			e.assignment, e.preemptionTargets = s.getAssignments(log, &e.Info, snap)
 			e.inadmissibleMsg = e.assignment.Message()
 			e.Info.LastAssignment = &e.assignment.LastState
-			if s.fairSharing.Enable && e.assignment.RepresentativeMode() != flavorassigner.NoFit {
-				e.dominantResourceShare = cq.DominantResourceShareWith(e.assignment.TotalRequestsFor(&w))
-			}
 		}
 		entries = append(entries, e)
 	}
@@ -383,16 +383,16 @@ func fits(cq *cache.ClusterQueueSnapshot, usage *workload.Usage, preemptedWorklo
 func resourcesToReserve(e *entry, cq *cache.ClusterQueueSnapshot) workload.Usage {
 	return workload.Usage{
 		Quota: quotaResourcesToReserve(e, cq),
-		TAS:   e.assignment.TASUsage(),
+		TAS:   e.assignment.Usage.TAS,
 	}
 }
 
 func quotaResourcesToReserve(e *entry, cq *cache.ClusterQueueSnapshot) resources.FlavorResourceQuantities {
 	if e.assignment.RepresentativeMode() != flavorassigner.Preempt {
-		return e.assignment.Usage
+		return e.assignment.Usage.Quota
 	}
 	reservedUsage := make(resources.FlavorResourceQuantities)
-	for fr, usage := range e.assignment.Usage {
+	for fr, usage := range e.assignment.Usage.Quota {
 		cqQuota := cq.QuotaFor(fr)
 		if e.assignment.Borrowing {
 			if cqQuota.BorrowingLimit == nil {
@@ -415,7 +415,7 @@ type partialAssignment struct {
 func (s *Scheduler) getAssignments(log logr.Logger, wl *workload.Info, snap *cache.Snapshot) (flavorassigner.Assignment, []*preemption.Target) {
 	assignment, targets := s.getInitialAssignments(log, wl, snap)
 	cq := snap.ClusterQueue(wl.ClusterQueue)
-	updateAssignmentForTAS(cq, wl, assignment, targets)
+	updateAssignmentForTAS(cq, wl, &assignment, targets)
 	return assignment, targets
 }
 
@@ -459,7 +459,7 @@ func (s *Scheduler) getInitialAssignments(log logr.Logger, wl *workload.Info, sn
 	return fullAssignment, nil
 }
 
-func updateAssignmentForTAS(cq *cache.ClusterQueueSnapshot, wl *workload.Info, assignment flavorassigner.Assignment, targets []*preemption.Target) {
+func updateAssignmentForTAS(cq *cache.ClusterQueueSnapshot, wl *workload.Info, assignment *flavorassigner.Assignment, targets []*preemption.Target) {
 	if features.Enabled(features.TopologyAwareScheduling) && assignment.RepresentativeMode() == flavorassigner.Preempt && wl.IsRequestingTAS() {
 		tasRequests := assignment.WorkloadsTopologyRequests(wl, cq)
 		var tasResult cache.TASAssignmentsResult
@@ -551,9 +551,8 @@ func (s *Scheduler) applyAdmissionWithSSA(ctx context.Context, w *kueue.Workload
 }
 
 type entryOrdering struct {
-	enableFairSharing bool
-	entries           []entry
-	workloadOrdering  workload.Ordering
+	entries          []entry
+	workloadOrdering workload.Ordering
 }
 
 func (e entryOrdering) Len() int {
@@ -576,12 +575,7 @@ func (e entryOrdering) Less(i, j int) bool {
 		return !aBorrows
 	}
 
-	// 2. Fair share, if enabled.
-	if e.enableFairSharing && a.dominantResourceShare != b.dominantResourceShare {
-		return a.dominantResourceShare < b.dominantResourceShare
-	}
-
-	// 3. Higher priority first if not disabled.
+	// 2. Higher priority first if not disabled.
 	if features.Enabled(features.PrioritySortingWithinCohort) {
 		p1 := priority.Priority(a.Obj)
 		p2 := priority.Priority(b.Obj)
@@ -590,7 +584,7 @@ func (e entryOrdering) Less(i, j int) bool {
 		}
 	}
 
-	// 4. FIFO.
+	// 3. FIFO.
 	aComparisonTimestamp := e.workloadOrdering.GetQueueOrderTimestamp(a.Obj)
 	bComparisonTimestamp := e.workloadOrdering.GetQueueOrderTimestamp(b.Obj)
 	return aComparisonTimestamp.Before(bComparisonTimestamp)
@@ -603,6 +597,13 @@ type entryIterator interface {
 	hasNext() bool
 }
 
+func makeIterator(ctx context.Context, entries []entry, workloadOrdering workload.Ordering, enableFairSharing bool) entryIterator {
+	if enableFairSharing {
+		return makeFairSharingIterator(ctx, entries, workloadOrdering)
+	}
+	return makeClassicalIterator(entries, workloadOrdering)
+}
+
 // classicalIterator returns entries ordered on:
 // 1. request under nominal quota before borrowing.
 // 2. fair sharing: lower DominantResourceShare first.
@@ -622,11 +623,10 @@ func (co *classicalIterator) pop() *entry {
 	return head
 }
 
-func makeIterator(entries []entry, workloadOrdering workload.Ordering, enableFairSharing bool) *classicalIterator {
+func makeClassicalIterator(entries []entry, workloadOrdering workload.Ordering) *classicalIterator {
 	sort.Sort(entryOrdering{
-		entries:           entries,
-		enableFairSharing: enableFairSharing,
-		workloadOrdering:  workloadOrdering,
+		entries:          entries,
+		workloadOrdering: workloadOrdering,
 	})
 	return &classicalIterator{
 		entries: entries,
diff --git a/site/content/en/docs/concepts/cluster_queue.md b/site/content/en/docs/concepts/cluster_queue.md
index 0a0738b36..dfe111a25 100644
--- a/site/content/en/docs/concepts/cluster_queue.md
+++ b/site/content/en/docs/concepts/cluster_queue.md
@@ -449,8 +449,9 @@ The fields above do the following:
     priority.
 
 - `borrowWithinCohort` determines whether a pending Workload can preempt
-  Workloads from other ClusterQueues if the workload requires borrowing. This
-  field requires to specify `policy` sub-field with possible values:
+  Workloads from other ClusterQueues if the workload requires borrowing.
+  May only be configured with Classical Preemption, and __not__ with Fair Sharing.
+  This field requires to specify `policy` sub-field with possible values:
   - `Never` (default): do not preempt Workloads in the cohort if borrowing is required.
   - `LowerPriority`: if the pending Workload requires borrowing, only preempt
     Workloads in the cohort that have lower priority than the pending Workload.
diff --git a/site/content/en/docs/concepts/preemption.md b/site/content/en/docs/concepts/preemption.md
index 63f21d8a9..cc78d5b20 100644
--- a/site/content/en/docs/concepts/preemption.md
+++ b/site/content/en/docs/concepts/preemption.md
@@ -119,8 +119,8 @@ admitted when accounting back the quota usage of the target Workload.
 
 Fair sharing introduces the concepts of ClusterQueue share values and preemption
 strategies. These work together with the preemption policies set in
-`withinClusterQueue` and `reclaimWithinCohort` to determine if a pending
-Workload can preempt an admitted Workload. Fair sharing uses preemptions to
+`withinClusterQueue` and `reclaimWithinCohort` (but __not__ `borrowWithinCohort`) to determine if a pending
+Workload can preempt an admitted Workload in Fair sharing. Fair sharing uses preemptions to
 achieve an equal or weighted share of the borrowable resources between the
 tenants of a cohort.
 
diff --git a/site/content/en/docs/reference/kueue.v1beta1.md b/site/content/en/docs/reference/kueue.v1beta1.md
index b6e6e95ea..6f06ebe80 100644
--- a/site/content/en/docs/reference/kueue.v1beta1.md
+++ b/site/content/en/docs/reference/kueue.v1beta1.md
@@ -568,7 +568,8 @@ If empty, the AdmissionCheck will run for all workloads submitted to the Cluster
 
 
 <p>BorrowWithinCohort contains configuration which allows to preempt workloads
-within cohort while borrowing.</p>
+within cohort while borrowing. It only works with Classical Preemption,
+<strong>not</strong> with Fair Sharing.</p>
 
 
 <table class="table">
@@ -704,6 +705,23 @@ in the cluster queue.</p>
 
 <p>ClusterQueuePreemption contains policies to preempt Workloads from this
 ClusterQueue or the ClusterQueue's cohort.</p>
+<p>Preemption may be configured to work in the following scenarios:</p>
+<ul>
+<li>When a Workload fits within the nominal quota of the ClusterQueue, but
+the quota is currently borrowed by other ClusterQueues in the cohort.
+We preempt workloads in other ClusterQueues to allow this ClusterQueue to
+reclaim its nominal quota. Configured using reclaimWithinCohort.</li>
+<li>When a Workload doesn't fit within the nominal quota of the ClusterQueue
+and there are admitted Workloads in the ClusterQueue with lower priority.
+Configured using withinClusterQueue.</li>
+<li>When a Workload may fit while both borrowing and preempting
+low priority workloads in the Cohort. Configured using borrowWithinCohort.</li>
+<li>When FairSharing is enabled, to maintain fair distribution of
+unused resources. See FairSharing documentation.</li>
+</ul>
+<p>The preemption algorithm tries to find a minimal set of Workloads to
+preempt to accomomdate the pending Workload, preempting Workloads with
+lower priority first.</p>
 
 
 <table class="table">
@@ -737,9 +755,7 @@ in the cohort that satisfy the fair sharing preemptionStrategies.</li>
 <a href="#kueue-x-k8s-io-v1beta1-BorrowWithinCohort"><code>BorrowWithinCohort</code></a>
 </td>
 <td>
-   <p>borrowWithinCohort provides configuration to allow preemption within
-cohort while borrowing.</p>
-</td>
+   <span class="text-muted">No description provided.</span></td>
 </tr>
 <tr><td><code>withinClusterQueue</code> <B>[Required]</B><br/>
 <a href="#kueue-x-k8s-io-v1beta1-PreemptionPolicy"><code>PreemptionPolicy</code></a>
@@ -862,21 +878,7 @@ before borrowing or preempting in the flavor being evaluated.</p>
 <a href="#kueue-x-k8s-io-v1beta1-ClusterQueuePreemption"><code>ClusterQueuePreemption</code></a>
 </td>
 <td>
-   <p>preemption describes policies to preempt Workloads from this ClusterQueue
-or the ClusterQueue's cohort.</p>
-<p>Preemption can happen in two scenarios:</p>
-<ul>
-<li>When a Workload fits within the nominal quota of the ClusterQueue, but
-the quota is currently borrowed by other ClusterQueues in the cohort.
-Preempting Workloads in other ClusterQueues allows this ClusterQueue to
-reclaim its nominal quota.</li>
-<li>When a Workload doesn't fit within the nominal quota of the ClusterQueue
-and there are admitted Workloads in the ClusterQueue with lower priority.</li>
-</ul>
-<p>The preemption algorithm tries to find a minimal set of Workloads to
-preempt to accommodate the pending Workload, preempting Workloads with
-lower priority first.</p>
-</td>
+   <span class="text-muted">No description provided.</span></td>
 </tr>
 <tr><td><code>admissionChecks</code><br/>
 <code>[]string</code>



Test patch:
diff --git a/pkg/cache/resource_test.go b/pkg/cache/resource_test.go
index b3ea2f7ac..f855f83f5 100644
--- a/pkg/cache/resource_test.go
+++ b/pkg/cache/resource_test.go
@@ -428,7 +428,7 @@ func TestAvailable(t *testing.T) {
 			// remove usage
 			{
 				for cqName, usage := range tc.usage {
-					snapshot.ClusterQueue(cqName).removeUsage(workload.Usage{Quota: usage})
+					snapshot.ClusterQueue(cqName).RemoveUsage(workload.Usage{Quota: usage})
 				}
 				clusterQueues := snapshot.ClusterQueues()
 				gotAvailable := make(map[string]resources.FlavorResourceQuantities, len(clusterQueues))
diff --git a/pkg/controller/jobs/deployment/deployment_webhook_test.go b/pkg/controller/jobs/deployment/deployment_webhook_test.go
index 0dbe8f98d..4ef5a992d 100644
--- a/pkg/controller/jobs/deployment/deployment_webhook_test.go
+++ b/pkg/controller/jobs/deployment/deployment_webhook_test.go
@@ -28,7 +28,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -53,7 +53,7 @@ func TestDefault(t *testing.T) {
 			want: testingdeployment.MakeDeployment("test-pod", "").
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 		},
 		"deployment with queue and pod template spec queue": {
@@ -64,7 +64,7 @@ func TestDefault(t *testing.T) {
 			want: testingdeployment.MakeDeployment("test-pod", "").
 				Queue("new-test-queue").
 				PodTemplateSpecQueue("new-test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 		},
 		"deployment without queue with pod template spec queue": {
@@ -78,7 +78,7 @@ func TestDefault(t *testing.T) {
 			want: testingdeployment.MakeDeployment("test-pod", "default").
 				Queue("default").
 				PodTemplateSpecQueue("default").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 		},
 		"LocalQueueDefaulting enabled, default lq is created, job has queue label": {
@@ -88,7 +88,7 @@ func TestDefault(t *testing.T) {
 			want: testingdeployment.MakeDeployment("test-pod", "").
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 		},
 		"LocalQueueDefaulting enabled, default lq isn't created, job doesn't have queue label": {
@@ -107,7 +107,7 @@ func TestDefault(t *testing.T) {
 				Queue("test-queue").
 				Label(constants.WorkloadPriorityClassLabel, "test").
 				PodTemplateSpecQueue("test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecLabel(constants.WorkloadPriorityClassLabel, "test").
 				Obj(),
 		},
@@ -122,7 +122,7 @@ func TestDefault(t *testing.T) {
 				Queue("new-test-queue").
 				Label(constants.WorkloadPriorityClassLabel, "new-test").
 				PodTemplateSpecQueue("new-test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecLabel(constants.WorkloadPriorityClassLabel, "new-test").
 				Obj(),
 		},
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
index a5e42e312..510658f3e 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
@@ -29,7 +29,7 @@ import (
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
-	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/util/testingjobs/leaderworkerset"
 	testingjobspod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
@@ -70,13 +70,13 @@ func TestPodReconciler(t *testing.T) {
 			lws: leaderworkerset.MakeLeaderWorkerSet("lws", "ns").Obj(),
 			pod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 			wantPod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 		},
 		"shouldn't set default values without queue name": {
@@ -85,14 +85,14 @@ func TestPodReconciler(t *testing.T) {
 			pod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
 				Label(leaderworkersetv1.GroupIndexLabelKey, "0").
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 			wantPod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
 				Label(leaderworkersetv1.GroupIndexLabelKey, "0").
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 		},
 		"should set default values": {
@@ -103,8 +103,8 @@ func TestPodReconciler(t *testing.T) {
 			pod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
 				Label(leaderworkersetv1.GroupIndexLabelKey, "0").
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 			wantPod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
@@ -113,9 +113,9 @@ func TestPodReconciler(t *testing.T) {
 				Group(GetWorkloadName(types.UID(testUID), "lws", "0")).
 				GroupTotalCount("1").
 				PrebuiltWorkload(GetWorkloadName(types.UID(testUID), "lws", "0")).
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
-				Annotation(podcontroller.RoleHashAnnotation, string(kueue.DefaultPodSetName)).
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
+				Annotation(podconstants.RoleHashAnnotation, string(kueue.DefaultPodSetName)).
 				Obj(),
 		},
 		"should set default values and priority class when has value": {
@@ -127,8 +127,8 @@ func TestPodReconciler(t *testing.T) {
 			pod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
 				Label(leaderworkersetv1.GroupIndexLabelKey, "0").
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 			wantPod: testingjobspod.MakePod("pod", "ns").
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
@@ -138,9 +138,9 @@ func TestPodReconciler(t *testing.T) {
 				Group(GetWorkloadName(types.UID(testUID), "lws", "0")).
 				GroupTotalCount("1").
 				PrebuiltWorkload(GetWorkloadName(types.UID(testUID), "lws", "0")).
-				Annotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podcontroller.GroupServingAnnotation, "true").
-				Annotation(podcontroller.RoleHashAnnotation, string(kueue.DefaultPodSetName)).
+				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
+				Annotation(podconstants.RoleHashAnnotation, string(kueue.DefaultPodSetName)).
 				Obj(),
 		},
 	}
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
index c70c64cb3..cc7cda28e 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
@@ -34,7 +34,7 @@ import (
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/util/testingjobs/leaderworkerset"
 )
@@ -67,7 +67,7 @@ func TestReconciler(t *testing.T) {
 			wantLeaderWorkerSet: leaderworkerset.MakeLeaderWorkerSet(testLWS, testNS).UID(testUID).Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
-					Annotation(podcontroller.IsGroupWorkloadAnnotationKey, podcontroller.IsGroupWorkloadAnnotationValue).
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
 						kueue.PodSet{
@@ -122,7 +122,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
-					Annotation(podcontroller.IsGroupWorkloadAnnotationKey, podcontroller.IsGroupWorkloadAnnotationValue).
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
 						kueue.PodSet{
@@ -192,7 +192,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
-					Annotation(podcontroller.IsGroupWorkloadAnnotationKey, podcontroller.IsGroupWorkloadAnnotationValue).
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
 						kueue.PodSet{
@@ -222,7 +222,7 @@ func TestReconciler(t *testing.T) {
 					).
 					Obj(),
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "1"), testNS).
-					Annotation(podcontroller.IsGroupWorkloadAnnotationKey, podcontroller.IsGroupWorkloadAnnotationValue).
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
 						kueue.PodSet{
@@ -334,7 +334,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
-					Annotation(podcontroller.IsGroupWorkloadAnnotationKey, podcontroller.IsGroupWorkloadAnnotationValue).
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
 						kueue.PodSet{
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
index 8f1042f0e..f44080766 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
@@ -33,7 +33,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -58,10 +58,10 @@ func TestDefault(t *testing.T) {
 			want: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "default").
 				LeaderTemplate(corev1.PodTemplateSpec{}).
 				Queue("default").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				LeaderTemplateSpecAnnotation(podcontroller.GroupServingAnnotation, "true").
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.GroupServingAnnotation, "true").
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 		},
 		"LocalQueueDefaulting enabled, default lq is created, job has queue label": {
@@ -74,10 +74,10 @@ func TestDefault(t *testing.T) {
 			want: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
 				LeaderTemplate(corev1.PodTemplateSpec{}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				LeaderTemplateSpecAnnotation(podcontroller.GroupServingAnnotation, "true").
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.GroupServingAnnotation, "true").
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
 				Obj(),
 		},
 		"LocalQueueDefaulting enabled, default lq isn't created, job doesn't have queue label": {
@@ -241,14 +241,14 @@ func TestValidateUpdate(t *testing.T) {
 			oldObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
 				LeaderTemplate(corev1.PodTemplateSpec{}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
 				LeaderTemplate(corev1.PodTemplateSpec{}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 		},
 		"change queue name": {
@@ -324,8 +324,8 @@ func TestValidateUpdate(t *testing.T) {
 					},
 				}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
 				LeaderTemplate(corev1.PodTemplateSpec{
@@ -365,8 +365,8 @@ func TestValidateUpdate(t *testing.T) {
 					},
 				}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 		},
 		"change resources in container": {
@@ -408,8 +408,8 @@ func TestValidateUpdate(t *testing.T) {
 					},
 				}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
 				LeaderTemplate(corev1.PodTemplateSpec{
@@ -457,8 +457,8 @@ func TestValidateUpdate(t *testing.T) {
 					},
 				}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 			wantErr: field.ErrorList{
 				&field.Error{
@@ -510,8 +510,8 @@ func TestValidateUpdate(t *testing.T) {
 					},
 				}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
 				LeaderTemplate(corev1.PodTemplateSpec{
@@ -559,8 +559,8 @@ func TestValidateUpdate(t *testing.T) {
 					},
 				}).
 				Queue("test-queue").
-				LeaderTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
-				WorkerTemplateSpecAnnotation(podcontroller.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				Obj(),
 			wantErr: field.ErrorList{
 				&field.Error{
diff --git a/pkg/controller/jobs/pod/pod_controller_test.go b/pkg/controller/jobs/pod/pod_controller_test.go
index 1f13b26e6..d1924c799 100644
--- a/pkg/controller/jobs/pod/pod_controller_test.go
+++ b/pkg/controller/jobs/pod/pod_controller_test.go
@@ -42,6 +42,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	controllerconsts "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/podset"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -346,7 +347,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("test-queue").
@@ -674,7 +675,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
@@ -682,7 +683,7 @@ func TestReconciler(t *testing.T) {
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
 					Annotations(map[string]string{
-						"kueue.x-k8s.io/is-group-workload":                           "true",
+						podconstants.IsGroupWorkloadAnnotationKey:                    podconstants.IsGroupWorkloadAnnotationValue,
 						controllerconsts.ProvReqAnnotationPrefix + "test-annotation": "test-val"}).
 					Obj(),
 			},
@@ -707,7 +708,7 @@ func TestReconciler(t *testing.T) {
 					Annotation("invalid-provreq-prefix/test-annotation-2", "test-val-2").
 					Group("test-group").
 					GroupTotalCount("3").
-					Annotation(GroupFastAdmissionAnnotation, "true").
+					Annotation(podconstants.GroupFastAdmissionAnnotationKey, podconstants.GroupFastAdmissionAnnotationValue).
 					Obj(),
 			},
 			wantPods: []corev1.Pod{
@@ -721,7 +722,7 @@ func TestReconciler(t *testing.T) {
 					Annotation("invalid-provreq-prefix/test-annotation-2", "test-val-2").
 					Group("test-group").
 					GroupTotalCount("3").
-					Annotation(GroupFastAdmissionAnnotation, "true").
+					Annotation(podconstants.GroupFastAdmissionAnnotationKey, podconstants.GroupFastAdmissionAnnotationValue).
 					Obj(),
 			},
 			wantWorkloads: []kueue.Workload{
@@ -729,14 +730,14 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 3).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
 					Priority(0).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					Annotations(map[string]string{
-						"kueue.x-k8s.io/is-group-workload":                           "true",
+						podconstants.IsGroupWorkloadAnnotationKey:                    podconstants.IsGroupWorkloadAnnotationValue,
 						controllerconsts.ProvReqAnnotationPrefix + "test-annotation": "test-val"}).
 					Obj(),
 			},
@@ -798,7 +799,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet("dc85db45", 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
@@ -806,7 +807,7 @@ func TestReconciler(t *testing.T) {
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
 					Annotations(map[string]string{
-						"kueue.x-k8s.io/is-group-workload": "true",
+						podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue,
 					}).
 					MaximumExecutionTimeSeconds(10).
 					Obj(),
@@ -848,14 +849,14 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet("dc85db45", 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
 					Priority(0).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
-					Annotations(map[string]string{"kueue.x-k8s.io/is-group-workload": "true"}).
+					Annotations(map[string]string{podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue}).
 					MaximumExecutionTimeSeconds(5).
 					Obj(),
 			},
@@ -885,14 +886,14 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet("dc85db45", 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
 					Priority(0).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
-					Annotations(map[string]string{"kueue.x-k8s.io/is-group-workload": "true"}).
+					Annotations(map[string]string{podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue}).
 					MaximumExecutionTimeSeconds(10).
 					Obj(),
 			},
@@ -950,7 +951,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
@@ -964,7 +965,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
@@ -2084,7 +2085,7 @@ func TestReconciler(t *testing.T) {
 					Priority(0).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
-					Annotations(map[string]string{"kueue.x-k8s.io/is-group-workload": "true"}).
+					Annotations(map[string]string{podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue}).
 					Obj(),
 			},
 			workloadCmpOpts: defaultWorkloadCmpOpts,
@@ -2331,7 +2332,7 @@ func TestReconciler(t *testing.T) {
 				*utiltesting.MakeWorkload("test-group", "ns").Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 2).
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Request(corev1.ResourceCPU, "1").
 							Obj(),
 					).
@@ -2339,7 +2340,7 @@ func TestReconciler(t *testing.T) {
 					Priority(0).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
-					Annotations(map[string]string{"kueue.x-k8s.io/is-group-workload": "true"}).
+					Annotations(map[string]string{podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue}).
 					Obj(),
 			},
 			workloadCmpOpts: defaultWorkloadCmpOpts,
@@ -2406,7 +2407,7 @@ func TestReconciler(t *testing.T) {
 			pods: []corev1.Pod{
 				*basePodWrapper.
 					Clone().
-					Annotation("kueue.x-k8s.io/retriable-in-group", "false").
+					Annotation(podconstants.RetriableInGroupAnnotationKey, podconstants.RetriableInGroupAnnotationValue).
 					ManagedByKueueLabel().
 					KueueFinalizer().
 					Group("test-group").
@@ -2436,7 +2437,7 @@ func TestReconciler(t *testing.T) {
 			wantPods: []corev1.Pod{
 				*basePodWrapper.
 					Clone().
-					Annotation("kueue.x-k8s.io/retriable-in-group", "false").
+					Annotation(podconstants.RetriableInGroupAnnotationKey, podconstants.RetriableInGroupAnnotationValue).
 					ManagedByKueueLabel().
 					Group("test-group").
 					GroupTotalCount("3").
@@ -2648,7 +2649,7 @@ func TestReconciler(t *testing.T) {
 					KueueFinalizer().
 					Group("test-group").
 					GroupTotalCount("3").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					StatusPhase(corev1.PodFailed).
 					Obj(),
 				*basePodWrapper.
@@ -2658,7 +2659,7 @@ func TestReconciler(t *testing.T) {
 					KueueFinalizer().
 					Group("test-group").
 					GroupTotalCount("3").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					StatusPhase(corev1.PodRunning).
 					Obj(),
 				*basePodWrapper.
@@ -2669,7 +2670,7 @@ func TestReconciler(t *testing.T) {
 					Group("test-group").
 					Request(corev1.ResourceMemory, "1Gi").
 					GroupTotalCount("3").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					StatusPhase(corev1.PodSucceeded).
 					Obj(),
 			},
@@ -2680,7 +2681,7 @@ func TestReconciler(t *testing.T) {
 					KueueFinalizer().
 					Group("test-group").
 					GroupTotalCount("3").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					StatusPhase(corev1.PodFailed).
 					Obj(),
 				*basePodWrapper.
@@ -2690,7 +2691,7 @@ func TestReconciler(t *testing.T) {
 					KueueFinalizer().
 					Group("test-group").
 					GroupTotalCount("3").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					StatusPhase(corev1.PodRunning).
 					Obj(),
 				*basePodWrapper.
@@ -2701,7 +2702,7 @@ func TestReconciler(t *testing.T) {
 					Group("test-group").
 					Request(corev1.ResourceMemory, "1Gi").
 					GroupTotalCount("3").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					StatusPhase(corev1.PodSucceeded).
 					Obj(),
 			},
@@ -2796,13 +2797,13 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 1).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
 					Priority(0).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
-					Annotations(map[string]string{"kueue.x-k8s.io/is-group-workload": "true"}).
+					Annotations(map[string]string{podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue}).
 					Obj(),
 			},
 			workloadCmpOpts: defaultWorkloadCmpOpts,
@@ -3238,14 +3239,14 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 1).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
 					Priority(0).
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
-					Annotations(map[string]string{"kueue.x-k8s.io/is-group-workload": "true"}).
+					Annotations(map[string]string{podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue}).
 					Obj(),
 			},
 			workloadCmpOpts: defaultWorkloadCmpOpts,
@@ -4069,7 +4070,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("test-queue").
@@ -4127,7 +4128,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
@@ -4135,7 +4136,7 @@ func TestReconciler(t *testing.T) {
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod", "test-uid").
 					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "pod2", "test-uid").
 					Annotations(map[string]string{
-						"kueue.x-k8s.io/is-group-workload": "true"}).
+						podconstants.IsGroupWorkloadAnnotationKey: podconstants.IsGroupWorkloadAnnotationValue}).
 					Labels(map[string]string{
 						"toCopyKey1": "toCopyValue1",
 						"toCopyKey2": "toCopyValue2",
@@ -4238,7 +4239,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
@@ -4268,7 +4269,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
@@ -4324,7 +4325,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
@@ -4351,7 +4352,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(
 						*utiltesting.MakePodSet(kueue.NewPodSetReference(podUID), 2).
 							Request(corev1.ResourceCPU, "1").
-							SchedulingGates(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}).
+							SchedulingGates(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}).
 							Obj(),
 					).
 					Queue("user-queue").
diff --git a/pkg/controller/jobs/pod/pod_webhook_test.go b/pkg/controller/jobs/pod/pod_webhook_test.go
index 42df48297..50da59804 100644
--- a/pkg/controller/jobs/pod/pod_webhook_test.go
+++ b/pkg/controller/jobs/pod/pod_webhook_test.go
@@ -34,6 +34,7 @@ import (
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -683,7 +684,7 @@ func TestValidateUpdate(t *testing.T) {
 			newPod: testingpod.MakePod("test-pod", "test-ns").
 				ManagedByKueueLabel().
 				OwnerReference("parent-job", batchv1.SchemeGroupVersion.WithKind("Job")).
-				Annotation(SuspendedByParentAnnotation, "job").
+				Annotation(podconstants.SuspendedByParentAnnotation, "job").
 				Obj(),
 		},
 		"pod with group name and no group total count": {
@@ -755,7 +756,7 @@ func TestValidateUpdate(t *testing.T) {
 			oldPod: testingpod.MakePod("test-pod", "test-ns").
 				Group("test-group").
 				GroupTotalCount("2").
-				Annotation("kueue.x-k8s.io/retriable-in-group", "false").
+				Annotation(podconstants.RetriableInGroupAnnotationKey, podconstants.RetriableInGroupAnnotationValue).
 				Obj(),
 			newPod: testingpod.MakePod("test-pod", "test-ns").
 				Group("test-group").
@@ -772,12 +773,12 @@ func TestValidateUpdate(t *testing.T) {
 			oldPod: testingpod.MakePod("test-pod", "test-ns").
 				Group("test-group").
 				GroupTotalCount("2").
-				Annotation("kueue.x-k8s.io/retriable-in-group", "false").
+				Annotation(podconstants.RetriableInGroupAnnotationKey, podconstants.RetriableInGroupAnnotationValue).
 				Obj(),
 			newPod: testingpod.MakePod("test-pod", "test-ns").
 				Group("test-group").
 				GroupTotalCount("2").
-				Annotation("kueue.x-k8s.io/retriable-in-group", "true").
+				Annotation(podconstants.RetriableInGroupAnnotationKey, "true").
 				Obj(),
 			wantErr: field.ErrorList{
 				&field.Error{
diff --git a/pkg/controller/jobs/statefulset/statefulset_reconciler_test.go b/pkg/controller/jobs/statefulset/statefulset_reconciler_test.go
index 7a116d7f0..4021af89d 100644
--- a/pkg/controller/jobs/statefulset/statefulset_reconciler_test.go
+++ b/pkg/controller/jobs/statefulset/statefulset_reconciler_test.go
@@ -27,7 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	testingjobspod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
 	statefulsettesting "sigs.k8s.io/kueue/pkg/util/testingjobs/statefulset"
@@ -53,31 +53,31 @@ func TestReconciler(t *testing.T) {
 			stsKey: client.ObjectKey{Name: "sts", Namespace: "ns"},
 			pods: []corev1.Pod{
 				*testingjobspod.MakePod("pod1", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					KueueFinalizer().
 					StatusPhase(corev1.PodSucceeded).
 					Obj(),
 				*testingjobspod.MakePod("pod2", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					KueueFinalizer().
 					StatusPhase(corev1.PodFailed).
 					Obj(),
 				*testingjobspod.MakePod("pod3", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					KueueFinalizer().
 					Obj(),
 			},
 			wantPods: []corev1.Pod{
 				*testingjobspod.MakePod("pod1", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					StatusPhase(corev1.PodSucceeded).
 					Obj(),
 				*testingjobspod.MakePod("pod2", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					StatusPhase(corev1.PodFailed).
 					Obj(),
 				*testingjobspod.MakePod("pod3", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					Obj(),
 			},
 		},
@@ -93,31 +93,31 @@ func TestReconciler(t *testing.T) {
 				DeepCopy(),
 			pods: []corev1.Pod{
 				*testingjobspod.MakePod("pod1", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					KueueFinalizer().
 					StatusPhase(corev1.PodSucceeded).
 					Obj(),
 				*testingjobspod.MakePod("pod2", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					KueueFinalizer().
 					StatusPhase(corev1.PodFailed).
 					Obj(),
 				*testingjobspod.MakePod("pod3", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					KueueFinalizer().
 					Obj(),
 			},
 			wantPods: []corev1.Pod{
 				*testingjobspod.MakePod("pod1", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					StatusPhase(corev1.PodSucceeded).
 					Obj(),
 				*testingjobspod.MakePod("pod2", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					StatusPhase(corev1.PodFailed).
 					Obj(),
 				*testingjobspod.MakePod("pod3", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					KueueFinalizer().
 					Obj(),
 			},
@@ -136,36 +136,36 @@ func TestReconciler(t *testing.T) {
 				DeepCopy(),
 			pods: []corev1.Pod{
 				*testingjobspod.MakePod("pod1", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					Label(appsv1.ControllerRevisionHashLabelKey, "1").
-					Gate(pod.SchedulingGateName).
+					Gate(podconstants.SchedulingGateName).
 					KueueFinalizer().
 					Obj(),
 				*testingjobspod.MakePod("pod2", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					Label(appsv1.ControllerRevisionHashLabelKey, "1").
 					KueueFinalizer().
 					Obj(),
 				*testingjobspod.MakePod("pod3", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					Label(appsv1.ControllerRevisionHashLabelKey, "2").
-					Gate(pod.SchedulingGateName).
+					Gate(podconstants.SchedulingGateName).
 					KueueFinalizer().
 					Obj(),
 			},
 			wantPods: []corev1.Pod{
 				*testingjobspod.MakePod("pod1", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					Label(appsv1.ControllerRevisionHashLabelKey, "1").
 					Obj(),
 				*testingjobspod.MakePod("pod2", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					Label(appsv1.ControllerRevisionHashLabelKey, "1").
 					Obj(),
 				*testingjobspod.MakePod("pod3", "ns").
-					Label(pod.GroupNameLabel, GetWorkloadName("sts")).
+					Label(podconstants.GroupNameLabel, GetWorkloadName("sts")).
 					Label(appsv1.ControllerRevisionHashLabelKey, "2").
-					Gate(pod.SchedulingGateName).
+					Gate(podconstants.SchedulingGateName).
 					KueueFinalizer().
 					Obj(),
 			},
diff --git a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
index e4fecce11..5be078fce 100644
--- a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
+++ b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
@@ -37,7 +37,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/leaderworkerset"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -63,11 +63,11 @@ func TestDefault(t *testing.T) {
 				Replicas(10).
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(10).
-				PodTemplateSpecPodGroupFastAdmissionAnnotation(true).
-				PodTemplateSpecPodGroupServingAnnotation(true).
+				PodTemplateSpecPodGroupFastAdmissionAnnotation().
+				PodTemplateSpecPodGroupServingAnnotation().
 				PodTemplateSpecPodGroupPodIndexLabelAnnotation(appsv1.PodIndexLabel).
 				Obj(),
 		},
@@ -83,12 +83,12 @@ func TestDefault(t *testing.T) {
 				Queue("test-queue").
 				Label(constants.WorkloadPriorityClassLabel, "test").
 				PodTemplateSpecQueue("test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecLabel(constants.WorkloadPriorityClassLabel, "test").
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(10).
-				PodTemplateSpecPodGroupFastAdmissionAnnotation(true).
-				PodTemplateSpecPodGroupServingAnnotation(true).
+				PodTemplateSpecPodGroupFastAdmissionAnnotation().
+				PodTemplateSpecPodGroupServingAnnotation().
 				PodTemplateSpecPodGroupPodIndexLabelAnnotation(appsv1.PodIndexLabel).
 				Obj(),
 		},
@@ -102,9 +102,9 @@ func TestDefault(t *testing.T) {
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(1).
 				PodTemplateSpecQueue("test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
-				PodTemplateSpecPodGroupFastAdmissionAnnotation(true).
-				PodTemplateSpecPodGroupServingAnnotation(true).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateSpecPodGroupFastAdmissionAnnotation().
+				PodTemplateSpecPodGroupServingAnnotation().
 				PodTemplateSpecPodGroupPodIndexLabelAnnotation(appsv1.PodIndexLabel).
 				Obj(),
 		},
@@ -115,11 +115,11 @@ func TestDefault(t *testing.T) {
 			want: testingstatefulset.MakeStatefulSet("test-pod", "default").
 				Queue("default").
 				PodTemplateSpecQueue("default").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(1).
-				PodTemplateSpecPodGroupFastAdmissionAnnotation(true).
-				PodTemplateSpecPodGroupServingAnnotation(true).
+				PodTemplateSpecPodGroupFastAdmissionAnnotation().
+				PodTemplateSpecPodGroupServingAnnotation().
 				PodTemplateSpecPodGroupPodIndexLabelAnnotation(appsv1.PodIndexLabel).
 				Obj(),
 		},
@@ -130,11 +130,11 @@ func TestDefault(t *testing.T) {
 			want: testingstatefulset.MakeStatefulSet("test-pod", "").
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
-				PodTemplateAnnotation(pod.SuspendedByParentAnnotation, FrameworkName).
+				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(1).
-				PodTemplateSpecPodGroupFastAdmissionAnnotation(true).
-				PodTemplateSpecPodGroupServingAnnotation(true).
+				PodTemplateSpecPodGroupFastAdmissionAnnotation().
+				PodTemplateSpecPodGroupServingAnnotation().
 				PodTemplateSpecPodGroupPodIndexLabelAnnotation(appsv1.PodIndexLabel).
 				Obj(),
 		},
@@ -235,8 +235,8 @@ func TestValidateUpdate(t *testing.T) {
 			oldObj: &appsv1.StatefulSet{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{
-						constants.QueueLabel: "queue1",
-						pod.GroupNameLabel:   "group1",
+						constants.QueueLabel:        "queue1",
+						podconstants.GroupNameLabel: "group1",
 					},
 				},
 				Spec: appsv1.StatefulSetSpec{
@@ -253,8 +253,8 @@ func TestValidateUpdate(t *testing.T) {
 			newObj: &appsv1.StatefulSet{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{
-						constants.QueueLabel: "queue1",
-						pod.GroupNameLabel:   "group1",
+						constants.QueueLabel:        "queue1",
+						podconstants.GroupNameLabel: "group1",
 					},
 				},
 				Spec: appsv1.StatefulSetSpec{
diff --git a/pkg/scheduler/flavorassigner/flavorassigner_test.go b/pkg/scheduler/flavorassigner/flavorassigner_test.go
index b9595e0eb..83f0f97d3 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner_test.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner_test.go
@@ -110,10 +110,10 @@ func TestAssignFlavors(t *testing.T) {
 						Count: 1,
 					},
 				},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "default", Resource: corev1.ResourceCPU}:    1_000,
 					{Flavor: "default", Resource: corev1.ResourceMemory}: utiltesting.Mi,
-				},
+				}},
 			},
 		},
 		"single flavor, fits tainted flavor": {
@@ -147,9 +147,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "tainted", Resource: corev1.ResourceCPU}: 1_000,
-				},
+				}},
 			},
 		},
 		"single flavor, fits tainted flavor with toleration": {
@@ -175,9 +175,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "taint_and_toleration", Resource: corev1.ResourceCPU}: 1_000,
-				},
+				}},
 			},
 		},
 		"single flavor, used resources, doesn't fit": {
@@ -210,9 +210,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "default", Resource: corev1.ResourceCPU}: 2_000,
-				},
+				}},
 			},
 		},
 		"multiple resource groups, fits": {
@@ -254,10 +254,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}:      3_000,
 					{Flavor: "b_one", Resource: corev1.ResourceMemory}: 10 * utiltesting.Mi,
-				},
+				}},
 			},
 		},
 		"multiple resource groups, one could fit with preemption, other doesn't fit": {
@@ -296,7 +296,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
 		"multiple resource groups with multiple resources, fits": {
@@ -342,11 +342,11 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}:    3_000,
 					{Flavor: "two", Resource: corev1.ResourceMemory}: 10 * utiltesting.Mi,
 					{Flavor: "b_one", Resource: "example.com/gpu"}:   3,
-				},
+				}},
 			},
 		},
 		"multiple resource groups with multiple resources, fits with different modes": {
@@ -411,11 +411,11 @@ func TestAssignFlavors(t *testing.T) {
 					Count: 1,
 				}},
 				Borrowing: true,
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}:    3_000,
 					{Flavor: "two", Resource: corev1.ResourceMemory}: 10 * utiltesting.Mi,
 					{Flavor: "b_one", Resource: "example.com/gpu"}:   3,
-				},
+				}},
 			},
 		},
 		"multiple resources in a group, doesn't fit": {
@@ -451,7 +451,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
 		"multiple flavors, fits while skipping tainted flavor": {
@@ -481,9 +481,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}: 3_000,
-				},
+				}},
 			},
 		},
 		"multiple flavors, fits a node selector": {
@@ -531,9 +531,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}: 1_000,
-				},
+				}},
 			},
 		},
 		"multiple flavors, fits with node affinity": {
@@ -585,10 +585,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}:    1_000,
 					{Flavor: "two", Resource: corev1.ResourceMemory}: utiltesting.Mi,
-				},
+				}},
 			},
 		},
 		"multiple flavors, node affinity fits any flavor": {
@@ -647,9 +647,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 1_000,
-				},
+				}},
 			},
 		},
 		"multiple flavors, doesn't fit node affinity": {
@@ -697,7 +697,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
 		"multiple specs, fit different flavors": {
@@ -743,10 +743,10 @@ func TestAssignFlavors(t *testing.T) {
 						Count: 1,
 					},
 				},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 3_000,
 					{Flavor: "two", Resource: corev1.ResourceCPU}: 5_000,
-				},
+				}},
 			},
 		},
 		"multiple specs, fits borrowing": {
@@ -806,10 +806,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 				},
 				Borrowing: true,
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "default", Resource: corev1.ResourceCPU}:    10_000,
 					{Flavor: "default", Resource: corev1.ResourceMemory}: 5 * utiltesting.Gi,
-				},
+				}},
 			},
 		},
 		"not enough space to borrow": {
@@ -846,7 +846,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
 		"past max, but can preempt in ClusterQueue": {
@@ -893,9 +893,9 @@ func TestAssignFlavors(t *testing.T) {
 					Count: 1,
 				}},
 				Borrowing: true,
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 2_000,
-				},
+				}},
 			},
 		},
 		"past min, but can preempt in ClusterQueue": {
@@ -928,9 +928,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 2_000,
-				},
+				}},
 			},
 		},
 		"past min, but can preempt in cohort and ClusterQueue": {
@@ -975,9 +975,9 @@ func TestAssignFlavors(t *testing.T) {
 					Count: 1,
 				}},
 				Borrowing: true,
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 2_000,
-				},
+				}},
 			},
 		},
 		"can only preempt flavors that match affinity": {
@@ -1019,9 +1019,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}: 2_000,
-				},
+				}},
 			},
 		},
 		"each podset requires preemption on a different flavor": {
@@ -1088,10 +1088,10 @@ func TestAssignFlavors(t *testing.T) {
 						Count: 10,
 					},
 				},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}:     2_000,
 					{Flavor: "tainted", Resource: corev1.ResourceCPU}: 10_000,
-				},
+				}},
 			},
 		},
 		"resource not listed in clusterQueue": {
@@ -1117,7 +1117,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
 		"num pods fit": {
@@ -1148,10 +1148,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 3,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "default", Resource: corev1.ResourcePods}: 3,
 					{Flavor: "default", Resource: corev1.ResourceCPU}:  3_000,
-				},
+				}},
 			},
 			wantRepMode: Fit,
 		},
@@ -1181,7 +1181,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 3,
 				}},
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
 		"with reclaimable pods": {
@@ -1217,10 +1217,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 3,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "default", Resource: corev1.ResourcePods}: 3,
 					{Flavor: "default", Resource: corev1.ResourceCPU}:  3_000,
-				},
+				}},
 			},
 			wantRepMode: Fit,
 		},
@@ -1262,10 +1262,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: "cpu"}:  9_000,
 					{Flavor: "one", Resource: "pods"}: 1,
-				},
+				}},
 			},
 		},
 		"preempt try next flavor": {
@@ -1302,10 +1302,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: "cpu"}:  9_000,
 					{Flavor: "two", Resource: "pods"}: 1,
-				},
+				}},
 			},
 		},
 		"borrow try next flavor, found the first flavor": {
@@ -1353,10 +1353,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}:  9_000,
 					{Flavor: "one", Resource: corev1.ResourcePods}: 1,
-				},
+				}},
 			},
 		},
 		"borrow try next flavor, found the second flavor": {
@@ -1404,10 +1404,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}:  9_000,
 					{Flavor: "two", Resource: corev1.ResourcePods}: 1,
-				},
+				}},
 			},
 		},
 		"borrow before try next flavor": {
@@ -1454,10 +1454,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: "cpu"}:  9_000,
 					{Flavor: "one", Resource: "pods"}: 1,
-				},
+				}},
 			},
 		},
 		"when borrowing while preemption is needed for flavor one; WhenCanBorrow=Borrow": {
@@ -1512,9 +1512,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 12_000,
-				},
+				}},
 			},
 		},
 		"when borrowing while preemption is needed for flavor one, no borrowingLimit; WhenCanBorrow=Borrow": {
@@ -1569,9 +1569,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 12_000,
-				},
+				}},
 			},
 		},
 		"when borrowing while preemption is needed for flavor one; WhenCanBorrow=TryNextFlavor": {
@@ -1620,9 +1620,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}: 12_000,
-				},
+				}},
 			},
 		},
 		"when borrowing while preemption is needed, but borrowingLimit exceeds the quota available in the cohort": {
@@ -1656,7 +1656,7 @@ func TestAssignFlavors(t *testing.T) {
 			},
 			wantRepMode: NoFit,
 			wantAssignment: Assignment{
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 				PodSets: []PodSetAssignment{
 					{
 						Name: kueue.DefaultPodSetName,
@@ -1718,10 +1718,10 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}:  9_000,
 					{Flavor: "two", Resource: corev1.ResourcePods}: 1,
-				},
+				}},
 			},
 		},
 		"lend try next flavor, found the first flavor": {
@@ -1769,10 +1769,10 @@ func TestAssignFlavors(t *testing.T) {
 					Count: 1,
 				}},
 				Borrowing: true,
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}:  9_000,
 					{Flavor: "one", Resource: corev1.ResourcePods}: 1,
-				},
+				}},
 			},
 		},
 		"quota exhausted, but can preempt in cohort and ClusterQueue": {
@@ -1819,10 +1819,10 @@ func TestAssignFlavors(t *testing.T) {
 					Count: 1,
 				}},
 				Borrowing: true,
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}:  9_000,
 					{Flavor: "one", Resource: corev1.ResourcePods}: 1,
-				},
+				}},
 			},
 		},
 		"when borrowing while preemption is needed for flavor one, fair sharing enabled, reclaimWithinCohort=Any": {
@@ -1870,9 +1870,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 12_000,
-				},
+				}},
 			},
 		},
 		"when borrowing while preemption is needed for flavor one, fair sharing enabled, reclaimWithinCohor=Never": {
@@ -1916,9 +1916,9 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}: 12_000,
-				},
+				}},
 			},
 		},
 	}
@@ -2185,9 +2185,9 @@ func TestDeletedFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "flavor", Resource: corev1.ResourceCPU}: 3_000,
-				},
+				}},
 			},
 		},
 		"flavor not found": {
@@ -2213,7 +2213,7 @@ func TestDeletedFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Usage: resources.FlavorResourceQuantities{},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
 	}
diff --git a/pkg/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index 1dd023257..e51ff02eb 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -2163,7 +2163,12 @@ func TestFairPreemptions(t *testing.T) {
 			incoming: utiltesting.MakeWorkload("a_incoming", "").Request(corev1.ResourceCPU, "4").Obj(),
 			targetCQ: "a",
 		},
-		"workloads under priority threshold can always be preempted": {
+		// preemption.borrowWithinCohort does not affect how
+		// we handle fair sharing preemptions. Lower priority
+		// workloads are not preempted unless
+		// DominantResourceShare value indicates that they
+		// should be preempted.
+		"workloads under priority threshold not capriciously preempted": {
 			clusterQueues: baseCQs,
 			admitted: []kueue.Workload{
 				*unitWl.Clone().Name("a1").SimpleReserveQuota("a", "default", now).Obj(),
@@ -2176,12 +2181,9 @@ func TestFairPreemptions(t *testing.T) {
 				*unitWl.Clone().Name("preemptible2").Priority(-3).SimpleReserveQuota("preemptible", "default", now).Obj(),
 				*unitWl.Clone().Name("preemptible3").Priority(-3).SimpleReserveQuota("preemptible", "default", now).Obj(),
 			},
-			incoming: utiltesting.MakeWorkload("a_incoming", "").Request(corev1.ResourceCPU, "2").Obj(),
-			targetCQ: "a",
-			wantPreempted: sets.New(
-				targetKeyReason("/preemptible1", kueue.InCohortFairSharingReason),
-				targetKeyReason("/preemptible2", kueue.InCohortReclaimWhileBorrowingReason),
-			),
+			incoming:      utiltesting.MakeWorkload("a_incoming", "").Request(corev1.ResourceCPU, "2").Obj(),
+			targetCQ:      "a",
+			wantPreempted: nil,
 		},
 		"preempt lower priority first, even if big": {
 			clusterQueues: baseCQs,
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index c248533fd..387769da1 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -254,6 +254,8 @@ func TestSchedule(t *testing.T) {
 		additionalClusterQueues []kueue.ClusterQueue
 		additionalLocalQueues   []kueue.LocalQueue
 
+		cohorts []kueuealpha.Cohort
+
 		// wantAssignments is a summary of all the admissions in the cache after this cycle.
 		wantAssignments map[string]kueue.Admission
 		// wantScheduled is the subset of workloads that got scheduled/admitted in this cycle.
@@ -1534,6 +1536,393 @@ func TestSchedule(t *testing.T) {
 				"eng-beta": {"eng-beta/older_new"},
 			},
 		},
+		// Cohort A provides 200 capacity, with 70 remaining.
+		// We denote Cohorts in UPPERCASE, and ClusterQueues
+		// in lowercase.
+		//
+		//            A
+		//        /       \
+		//       /          \
+		//     B(30)         C(100)
+		//    /    \        /  \
+		//   d(10) e(20)   f(0)  g(100)
+		//
+		// In (), we display current admissions.  These
+		// numbers are proportional to DominantResourceShare,
+		// which we call below pDRS.
+		//
+		// pending workloads -> resulting pDRS if admitted.
+		// d1: 70 -> d(80) , B(100)
+		// e1: 61 -> e(81) , B(91)
+		// f1:  1 -> f(1)  , C(101)
+		// g1:  1 -> g(101), C(101)
+		//
+		// We expect d1 to admit, since after its admission B
+		// has lower pDRS (100) than C (101) after admission
+		// of either f1 or g1.
+		//
+		// Though admission of e1 would result in an even
+		// lower pDRS of B (91), d1 won the tournament at the
+		// lower level, which we see by comparing d and e's
+		// pDRSs, 80 and 81 respectively, after admission of
+		// d1 and e1 respectively.
+		"hierarchical fair sharing schedule workload which wins tournament": {
+			enableFairSharing: true,
+			cohorts: []kueuealpha.Cohort{
+				utiltesting.MakeCohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "200").Obj(),
+					).Cohort,
+				utiltesting.MakeCohort("B").Parent("A").Cohort,
+				utiltesting.MakeCohort("C").Parent("A").Cohort,
+			},
+			additionalClusterQueues: []kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("d").
+					Cohort("B").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("e").
+					Cohort("B").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("f").
+					Cohort("C").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("g").
+					Cohort("C").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+			},
+			additionalLocalQueues: []kueue.LocalQueue{
+				*utiltesting.MakeLocalQueue("lq-d", "eng-alpha").ClusterQueue("d").Obj(),
+				*utiltesting.MakeLocalQueue("lq-e", "eng-alpha").ClusterQueue("e").Obj(),
+				*utiltesting.MakeLocalQueue("lq-f", "eng-alpha").ClusterQueue("f").Obj(),
+				*utiltesting.MakeLocalQueue("lq-g", "eng-alpha").ClusterQueue("g").Obj(),
+			},
+			workloads: []kueue.Workload{
+				utiltesting.MakeWorkload("d0", "eng-alpha").
+					Queue("lq-d").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					ReserveQuota(utiltesting.MakeAdmission("d", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj()).
+					Workload,
+				utiltesting.MakeWorkload("e0", "eng-alpha").
+					Queue("lq-e").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "20").
+						PodSet).
+					ReserveQuota(utiltesting.MakeAdmission("e", "one").Assignment(corev1.ResourceCPU, "on-demand", "20").Obj()).
+					Workload,
+				utiltesting.MakeWorkload("g0", "eng-alpha").
+					Queue("lq-g").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "100").
+						PodSet).
+					ReserveQuota(utiltesting.MakeAdmission("g", "one").Assignment(corev1.ResourceCPU, "on-demand", "100").Obj()).
+					Workload,
+				utiltesting.MakeWorkload("d1", "eng-alpha").
+					Queue("lq-d").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "70").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("e1", "eng-alpha").
+					Queue("lq-e").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "61").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("f1", "eng-alpha").
+					Queue("lq-f").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "1").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("g1", "eng-alpha").
+					Queue("lq-g").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "1").
+						PodSet).
+					Workload,
+			},
+			wantAssignments: map[string]kueue.Admission{
+				"eng-alpha/d0": *utiltesting.MakeAdmission("d", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj(),
+				"eng-alpha/e0": *utiltesting.MakeAdmission("e", "one").Assignment(corev1.ResourceCPU, "on-demand", "20").Obj(),
+				"eng-alpha/g0": *utiltesting.MakeAdmission("g", "one").Assignment(corev1.ResourceCPU, "on-demand", "100").Obj(),
+				"eng-alpha/d1": *utiltesting.MakeAdmission("d", "one").Assignment(corev1.ResourceCPU, "on-demand", "70").Obj(),
+			},
+			wantScheduled: []string{"eng-alpha/d1"},
+			wantLeft: map[string][]string{
+				"e": {"eng-alpha/e1"},
+				"g": {"eng-alpha/g1"},
+				"f": {"eng-alpha/f1"},
+			},
+		},
+		// b0 is already admitted, using 10 capacity.
+		// b1 - 50 capacity, and c1 - 75 capacity are pending.
+		//
+		// we expect b1 to schedule, as b0 + b1 = 60, is less than
+		// c1 = 75.
+		"fair sharing schedule workload with lowest drf after admission": {
+			enableFairSharing: true,
+			cohorts: []kueuealpha.Cohort{
+				utiltesting.MakeCohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "100").Obj(),
+					).Cohort,
+			},
+			additionalClusterQueues: []kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("b").
+					Cohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("c").
+					Cohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+			},
+			additionalLocalQueues: []kueue.LocalQueue{
+				*utiltesting.MakeLocalQueue("lq-b", "eng-alpha").ClusterQueue("b").Obj(),
+				*utiltesting.MakeLocalQueue("lq-c", "eng-alpha").ClusterQueue("c").Obj(),
+			},
+			workloads: []kueue.Workload{
+				utiltesting.MakeWorkload("b0", "eng-alpha").
+					Queue("lq-b").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					ReserveQuota(utiltesting.MakeAdmission("b", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj()).
+					Workload,
+				utiltesting.MakeWorkload("b1", "eng-alpha").
+					Queue("lq-b").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "50").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("c1", "eng-alpha").
+					Queue("lq-c").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "75").
+						PodSet).
+					Workload,
+			},
+			wantAssignments: map[string]kueue.Admission{
+				"eng-alpha/b0": *utiltesting.MakeAdmission("b", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj(),
+				"eng-alpha/b1": *utiltesting.MakeAdmission("b", "one").Assignment(corev1.ResourceCPU, "on-demand", "50").Obj(),
+			},
+			wantScheduled: []string{"eng-alpha/b1"},
+			wantLeft: map[string][]string{
+				"c": {"eng-alpha/c1"},
+			},
+		},
+		// Cohort A has Clusterqueue a, and capacity is
+		// provided by Cohort.
+		//
+		// Cohort B has Clusterqueue b, and capacity is
+		// provided by ClusterQueue.
+		//
+		// Clusterqueue c has no Cohort, and provides its own
+		// capacity.
+		//
+		// We ensure that all 3 pending workloads, one for each
+		// cq, schedules.
+		"fair sharing schedule singleton cqs and cq without cohort": {
+			enableFairSharing: true,
+			cohorts: []kueuealpha.Cohort{
+				utiltesting.MakeCohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "10").Obj(),
+					).Cohort,
+				utiltesting.MakeCohort("B").Cohort,
+			},
+			additionalClusterQueues: []kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("a").
+					Cohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("b").
+					Cohort("B").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "10").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("c").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "10").Obj(),
+					).
+					ClusterQueue,
+			},
+			additionalLocalQueues: []kueue.LocalQueue{
+				*utiltesting.MakeLocalQueue("lq-a", "eng-alpha").ClusterQueue("a").Obj(),
+				*utiltesting.MakeLocalQueue("lq-b", "eng-alpha").ClusterQueue("b").Obj(),
+				*utiltesting.MakeLocalQueue("lq-c", "eng-alpha").ClusterQueue("c").Obj(),
+			},
+			workloads: []kueue.Workload{
+				utiltesting.MakeWorkload("a1", "eng-alpha").
+					Queue("lq-a").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("b1", "eng-alpha").
+					Queue("lq-b").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("c1", "eng-alpha").
+					Queue("lq-c").
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					Workload,
+			},
+			wantAssignments: map[string]kueue.Admission{
+				"eng-alpha/a1": *utiltesting.MakeAdmission("a", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj(),
+				"eng-alpha/b1": *utiltesting.MakeAdmission("b", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj(),
+				"eng-alpha/c1": *utiltesting.MakeAdmission("c", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj(),
+			},
+			wantScheduled: []string{"eng-alpha/a1", "eng-alpha/b1", "eng-alpha/c1"},
+			wantLeft:      nil,
+		},
+		"fair sharing schedule highest priority first": {
+			enableFairSharing: true,
+			cohorts: []kueuealpha.Cohort{
+				utiltesting.MakeCohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "10").Obj(),
+					).Cohort,
+				utiltesting.MakeCohort("B").Cohort,
+			},
+			additionalClusterQueues: []kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("b").
+					Cohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("c").
+					Cohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+			},
+			additionalLocalQueues: []kueue.LocalQueue{
+				*utiltesting.MakeLocalQueue("lq-b", "eng-alpha").ClusterQueue("b").Obj(),
+				*utiltesting.MakeLocalQueue("lq-c", "eng-alpha").ClusterQueue("c").Obj(),
+			},
+			workloads: []kueue.Workload{
+				utiltesting.MakeWorkload("b1", "eng-alpha").
+					Queue("lq-b").
+					Priority(99).
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("c1", "eng-alpha").
+					Queue("lq-c").
+					Priority(101).
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					Workload,
+			},
+			wantAssignments: map[string]kueue.Admission{
+				"eng-alpha/c1": *utiltesting.MakeAdmission("c", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj(),
+			},
+			wantScheduled: []string{"eng-alpha/c1"},
+			wantLeft: map[string][]string{
+				"b": {"eng-alpha/b1"},
+			},
+		},
+		"fair sharing schedule earliest timestamp first": {
+			enableFairSharing: true,
+			cohorts: []kueuealpha.Cohort{
+				utiltesting.MakeCohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "10").Obj(),
+					).Cohort,
+				utiltesting.MakeCohort("B").Cohort,
+			},
+			additionalClusterQueues: []kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("b").
+					Cohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("c").
+					Cohort("A").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("on-demand").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+			},
+			additionalLocalQueues: []kueue.LocalQueue{
+				*utiltesting.MakeLocalQueue("lq-b", "eng-alpha").ClusterQueue("b").Obj(),
+				*utiltesting.MakeLocalQueue("lq-c", "eng-alpha").ClusterQueue("c").Obj(),
+			},
+			workloads: []kueue.Workload{
+				utiltesting.MakeWorkload("b1", "eng-alpha").
+					Creation(now.Add(time.Second)).
+					Queue("lq-b").
+					Priority(101).
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("c1", "eng-alpha").
+					Creation(now).
+					Queue("lq-c").
+					Priority(101).
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "10").
+						PodSet).
+					Workload,
+			},
+			wantAssignments: map[string]kueue.Admission{
+				"eng-alpha/c1": *utiltesting.MakeAdmission("c", "one").Assignment(corev1.ResourceCPU, "on-demand", "10").Obj(),
+			},
+			wantScheduled: []string{"eng-alpha/c1"},
+			wantLeft: map[string][]string{
+				"b": {"eng-alpha/b1"},
+			},
+		},
 		"minimal preemptions when target queue is exhausted": {
 			additionalClusterQueues: []kueue.ClusterQueue{
 				*utiltesting.MakeClusterQueue("other-alpha").
@@ -2607,6 +2996,13 @@ func TestSchedule(t *testing.T) {
 					t.Errorf("couldn't create the cluster queue: %v", err)
 				}
 			}
+
+			for _, cohort := range tc.cohorts {
+				if err := cqCache.AddOrUpdateCohort(&cohort); err != nil {
+					t.Fatalf("Inserting Cohort %s in cache: %v", cohort.Name, err)
+				}
+			}
+
 			scheduler := New(qManager, cqCache, cl, recorder, WithFairSharing(&config.FairSharing{Enable: tc.enableFairSharing}), WithClock(t, fakeClock))
 			gotScheduled := make(map[string]kueue.Admission)
 			var mu sync.Mutex
@@ -2963,7 +3359,7 @@ func TestEntryOrdering(t *testing.T) {
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			features.SetFeatureGateDuringTest(t, features.PrioritySortingWithinCohort, tc.prioritySorting)
-			iter := makeIterator(tc.input, tc.workloadOrdering, false)
+			iter := makeIterator(context.Background(), tc.input, tc.workloadOrdering, false)
 			order := make([]string, len(tc.input))
 			for i := range tc.input {
 				order[i] = iter.pop().Obj.Name
@@ -3787,7 +4183,7 @@ func TestResourcesToReserve(t *testing.T) {
 					},
 				},
 				Borrowing: tc.borrowing,
-				Usage:     tc.assignmentUsage,
+				Usage:     workload.Usage{Quota: tc.assignmentUsage},
 			}
 			e := &entry{assignment: assignment}
 			cl := utiltesting.NewClientBuilder().
diff --git a/pkg/util/testingjobs/leaderworkerset/wrappers.go b/pkg/util/testingjobs/leaderworkerset/wrappers.go
index 2ee1485bb..646ea989d 100644
--- a/pkg/util/testingjobs/leaderworkerset/wrappers.go
+++ b/pkg/util/testingjobs/leaderworkerset/wrappers.go
@@ -18,7 +18,6 @@ package leaderworkerset
 
 import (
 	"fmt"
-	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -30,7 +29,7 @@ import (
 
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 )
 
 // LeaderWorkerSetWrapper wraps a LeaderWorkerSet.
@@ -156,15 +155,11 @@ func (w *LeaderWorkerSetWrapper) WorkerTemplateSpecPodGroupNameLabel(
 	ownerName string, ownerUID types.UID, ownerGVK schema.GroupVersionKind,
 ) *LeaderWorkerSetWrapper {
 	gvk := jobframework.GetWorkloadNameForOwnerWithGVK(ownerName, ownerUID, ownerGVK)
-	return w.WorkerTemplateSpecLabel(pod.GroupNameLabel, gvk)
+	return w.WorkerTemplateSpecLabel(podconstants.GroupNameLabel, gvk)
 }
 
 func (w *LeaderWorkerSetWrapper) WorkerTemplateSpecPodGroupTotalCountAnnotation(replicas int32) *LeaderWorkerSetWrapper {
-	return w.WorkerTemplateSpecAnnotation(pod.GroupTotalCountAnnotation, fmt.Sprint(replicas))
-}
-
-func (w *LeaderWorkerSetWrapper) WorkerTemplateSpecPodGroupFastAdmissionAnnotation(enabled bool) *LeaderWorkerSetWrapper {
-	return w.WorkerTemplateSpecAnnotation(pod.GroupFastAdmissionAnnotation, strconv.FormatBool(enabled))
+	return w.WorkerTemplateSpecAnnotation(podconstants.GroupTotalCountAnnotation, fmt.Sprint(replicas))
 }
 
 func (w *LeaderWorkerSetWrapper) Image(image string, args []string) *LeaderWorkerSetWrapper {
diff --git a/pkg/util/testingjobs/pod/wrappers.go b/pkg/util/testingjobs/pod/wrappers.go
index 42227ce37..81b739aad 100644
--- a/pkg/util/testingjobs/pod/wrappers.go
+++ b/pkg/util/testingjobs/pod/wrappers.go
@@ -31,6 +31,7 @@ import (
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	controllerconsts "sigs.k8s.io/kueue/pkg/controller/constants"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 )
 
 // PodWrapper wraps a Pod.
@@ -124,12 +125,12 @@ func (p *PodWrapper) Namespace(n string) *PodWrapper {
 
 // Group updates the pod.GroupNameLabel of the Pod
 func (p *PodWrapper) Group(g string) *PodWrapper {
-	return p.Label("kueue.x-k8s.io/pod-group-name", g)
+	return p.Label(podconstants.GroupNameLabel, g)
 }
 
 // GroupTotalCount updates the pod.GroupTotalCountAnnotation of the Pod
 func (p *PodWrapper) GroupTotalCount(gtc string) *PodWrapper {
-	return p.Annotation("kueue.x-k8s.io/pod-group-total-count", gtc)
+	return p.Annotation(podconstants.GroupTotalCountAnnotation, gtc)
 }
 
 // GroupIndex updates the pod.GroupIndexLabel of the Pod
@@ -155,18 +156,18 @@ func (p *PodWrapper) ManagedByKueueLabel() *PodWrapper {
 	return p.Label(constants.ManagedByKueueLabelKey, constants.ManagedByKueueLabelValue)
 }
 
-func (p *PodWrapper) PodGroupServingAnnotation(enabled bool) *PodWrapper {
-	return p.Annotation("kueue.x-k8s.io/pod-group-serving", strconv.FormatBool(enabled))
+func (p *PodWrapper) PodGroupServingAnnotation() *PodWrapper {
+	return p.Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue)
 }
 
 // RoleHash updates the pod.RoleHashAnnotation of the pod
 func (p *PodWrapper) RoleHash(h string) *PodWrapper {
-	return p.Annotation("kueue.x-k8s.io/role-hash", h)
+	return p.Annotation(podconstants.RoleHashAnnotation, h)
 }
 
 // KueueSchedulingGate adds kueue scheduling gate to the Pod
 func (p *PodWrapper) KueueSchedulingGate() *PodWrapper {
-	return p.Gate("kueue.x-k8s.io/admission")
+	return p.Gate(podconstants.SchedulingGateName)
 }
 
 // TopologySchedulingGate adds kueue scheduling gate to the Pod
diff --git a/pkg/util/testingjobs/statefulset/wrappers.go b/pkg/util/testingjobs/statefulset/wrappers.go
index 9cd96ac16..ee5efae70 100644
--- a/pkg/util/testingjobs/statefulset/wrappers.go
+++ b/pkg/util/testingjobs/statefulset/wrappers.go
@@ -18,7 +18,6 @@ package statefulset
 
 import (
 	"fmt"
-	"strconv"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -30,7 +29,7 @@ import (
 	"sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 )
 
 // StatefulSetWrapper wraps a StatefulSet.
@@ -169,19 +168,19 @@ func (ss *StatefulSetWrapper) PodTemplateSpecPodGroupNameLabel(
 	ownerName string, ownerUID types.UID, ownerGVK schema.GroupVersionKind,
 ) *StatefulSetWrapper {
 	gvk := jobframework.GetWorkloadNameForOwnerWithGVK(ownerName, ownerUID, ownerGVK)
-	return ss.PodTemplateSpecLabel(pod.GroupNameLabel, gvk)
+	return ss.PodTemplateSpecLabel(podconstants.GroupNameLabel, gvk)
 }
 
 func (ss *StatefulSetWrapper) PodTemplateSpecPodGroupTotalCountAnnotation(replicas int32) *StatefulSetWrapper {
-	return ss.PodTemplateSpecAnnotation(pod.GroupTotalCountAnnotation, fmt.Sprint(replicas))
+	return ss.PodTemplateSpecAnnotation(podconstants.GroupTotalCountAnnotation, fmt.Sprint(replicas))
 }
 
-func (ss *StatefulSetWrapper) PodTemplateSpecPodGroupFastAdmissionAnnotation(enabled bool) *StatefulSetWrapper {
-	return ss.PodTemplateSpecAnnotation(pod.GroupFastAdmissionAnnotation, strconv.FormatBool(enabled))
+func (ss *StatefulSetWrapper) PodTemplateSpecPodGroupFastAdmissionAnnotation() *StatefulSetWrapper {
+	return ss.PodTemplateSpecAnnotation(podconstants.GroupFastAdmissionAnnotationKey, podconstants.GroupFastAdmissionAnnotationValue)
 }
 
-func (ss *StatefulSetWrapper) PodTemplateSpecPodGroupServingAnnotation(enabled bool) *StatefulSetWrapper {
-	return ss.PodTemplateSpecAnnotation(pod.GroupServingAnnotation, strconv.FormatBool(enabled))
+func (ss *StatefulSetWrapper) PodTemplateSpecPodGroupServingAnnotation() *StatefulSetWrapper {
+	return ss.PodTemplateSpecAnnotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue)
 }
 
 func (ss *StatefulSetWrapper) PodTemplateSpecPodGroupPodIndexLabelAnnotation(labelName string) *StatefulSetWrapper {
diff --git a/test/e2e/customconfigs/skipjobswithoutqueuename_test.go b/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
index 3abaccf6d..63832d94c 100644
--- a/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
+++ b/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
@@ -33,7 +33,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
-	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	awtesting "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
 	testingdeploy "sigs.k8s.io/kueue/pkg/util/testingjobs/deployment"
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index 098a02729..e06e1a726 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -46,6 +46,7 @@ import (
 	workloadpytorchjob "sigs.k8s.io/kueue/pkg/controller/jobs/kubeflow/jobs/pytorchjob"
 	workloadmpijob "sigs.k8s.io/kueue/pkg/controller/jobs/mpijob"
 	workloadpod "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	workloadraycluster "sigs.k8s.io/kueue/pkg/controller/jobs/raycluster"
 	workloadrayjob "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
@@ -208,7 +209,7 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 				Request("memory", "2G").
 				Queue(managerLq.Name).
 				Obj()
-				// Since it requires 2G of memory, this pod can only be admitted in worker 2.
+			// Since it requires 2G of memory, this pod can only be admitted in worker 2.
 
 			ginkgo.By("Creating the pod", func() {
 				gomega.Expect(k8sManagerClient.Create(ctx, pod)).Should(gomega.Succeed())
@@ -235,7 +236,7 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 				gomega.Eventually(func(g gomega.Gomega) {
 					createdPod := &corev1.Pod{}
 					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(pod), createdPod)).To(gomega.Succeed())
-					g.Expect(utilpod.HasGate(createdPod, "kueue.x-k8s.io/admission")).To(gomega.BeTrue())
+					g.Expect(utilpod.HasGate(createdPod, podconstants.SchedulingGateName)).To(gomega.BeTrue())
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 			finishReason := "PodCompleted"
diff --git a/test/e2e/singlecluster/deployment_test.go b/test/e2e/singlecluster/deployment_test.go
index 9ef2e5fc1..7c4bbd4a3 100644
--- a/test/e2e/singlecluster/deployment_test.go
+++ b/test/e2e/singlecluster/deployment_test.go
@@ -28,6 +28,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	deploymenttesting "sigs.k8s.io/kueue/pkg/util/testingjobs/deployment"
@@ -160,7 +161,7 @@ var _ = ginkgo.Describe("Deployment", func() {
 		createdWorkloads := make([]*kueue.Workload, 0, len(pods.Items))
 		ginkgo.By("Check that workloads are created but not admitted", func() {
 			for _, p := range pods.Items {
-				gomega.Expect(utilpod.HasGate(&p, pod.SchedulingGateName)).Should(gomega.BeTrue())
+				gomega.Expect(utilpod.HasGate(&p, podconstants.SchedulingGateName)).Should(gomega.BeTrue())
 				createdWorkload := &kueue.Workload{}
 				wlLookupKey := types.NamespacedName{
 					Name:      pod.GetWorkloadNameForPod(p.Name, p.UID),
diff --git a/test/e2e/singlecluster/pod_test.go b/test/e2e/singlecluster/pod_test.go
index 810263ce0..9a14bc6c6 100644
--- a/test/e2e/singlecluster/pod_test.go
+++ b/test/e2e/singlecluster/pod_test.go
@@ -23,12 +23,14 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	podtesting "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
 	"sigs.k8s.io/kueue/test/util"
@@ -90,7 +92,7 @@ var _ = ginkgo.Describe("Pod groups", func() {
 				gomega.Expect(k8sClient.Create(ctx, p)).To(gomega.Succeed())
 				gomega.Expect(p.Spec.SchedulingGates).
 					To(gomega.ContainElement(corev1.PodSchedulingGate{
-						Name: pod.SchedulingGateName}))
+						Name: podconstants.SchedulingGateName}))
 			}
 			ginkgo.By("Starting admission", func() {
 				// Verify that the Pods start with the appropriate selector.
@@ -138,7 +140,7 @@ var _ = ginkgo.Describe("Pod groups", func() {
 					for _, origPod := range group[:2] {
 						var p corev1.Pod
 						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(origPod), &p)).To(gomega.Succeed())
-						g.Expect(p.Spec.SchedulingGates).To(gomega.ContainElement(corev1.PodSchedulingGate{Name: pod.SchedulingGateName}))
+						g.Expect(p.Spec.SchedulingGates).To(gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}))
 					}
 				}, util.ConsistentDuration, util.Interval).Should(gomega.Succeed())
 			})
@@ -518,7 +520,12 @@ var _ = ginkgo.Describe("Pod groups", func() {
 			})
 
 			ginkgo.By("Call high priority group pods to complete", func() {
-				util.WaitForActivePodsAndTerminate(ctx, k8sClient, restClient, cfg, ns.Name, 2, 0)
+				selector, err := labels.Parse("kueue.x-k8s.io/pod-group-name=high-priority-group")
+				gomega.Expect(err).NotTo(gomega.HaveOccurred())
+				listOpts := &client.ListOptions{
+					LabelSelector: selector,
+				}
+				util.WaitForActivePodsAndTerminate(ctx, k8sClient, restClient, cfg, ns.Name, 2, 0, listOpts)
 			})
 
 			ginkgo.By("Verify the high priority group completes", func() {
diff --git a/test/e2e/singlecluster/tas_test.go b/test/e2e/singlecluster/tas_test.go
index 75f21039c..05664399f 100644
--- a/test/e2e/singlecluster/tas_test.go
+++ b/test/e2e/singlecluster/tas_test.go
@@ -32,6 +32,7 @@ import (
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/jobset"
 	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingjobset "sigs.k8s.io/kueue/pkg/util/testingjobs/jobset"
@@ -336,7 +337,7 @@ var _ = ginkgo.Describe("TopologyAwareScheduling", func() {
 			ginkgo.By("Creating the Pod", func() {
 				gomega.Expect(k8sClient.Create(ctx, p)).Should(gomega.Succeed())
 				gomega.Expect(p.Spec.SchedulingGates).To(gomega.ContainElements(
-					corev1.PodSchedulingGate{Name: podcontroller.SchedulingGateName},
+					corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName},
 					corev1.PodSchedulingGate{Name: kueuealpha.TopologySchedulingGate},
 				))
 				gomega.Expect(p.Labels).To(gomega.HaveKeyWithValue(kueuealpha.TASLabel, "true"))
@@ -396,7 +397,7 @@ var _ = ginkgo.Describe("TopologyAwareScheduling", func() {
 				for _, p := range group {
 					gomega.Expect(k8sClient.Create(ctx, p)).To(gomega.Succeed())
 					gomega.Expect(p.Spec.SchedulingGates).To(gomega.ContainElements(
-						corev1.PodSchedulingGate{Name: podcontroller.SchedulingGateName},
+						corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName},
 						corev1.PodSchedulingGate{Name: kueuealpha.TopologySchedulingGate},
 					))
 					gomega.Expect(p.Labels).To(gomega.HaveKeyWithValue(kueuealpha.TASLabel, "true"))
diff --git a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
index 880763ed2..0b542bdab 100644
--- a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
@@ -40,6 +40,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	testingnode "sigs.k8s.io/kueue/pkg/util/testingjobs/node"
@@ -153,7 +154,7 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 				gomega.Expect(createdPod.Spec.SchedulingGates).To(
-					gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+					gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 					"Pod should have scheduling gate",
 				)
 
@@ -370,7 +371,7 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 					}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 					gomega.Expect(createdPod.Spec.SchedulingGates).NotTo(
-						gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+						gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 						"Pod shouldn't have scheduling gate",
 					)
 
@@ -448,7 +449,7 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 						gomega.Eventually(func(g gomega.Gomega) {
 							g.Expect(k8sClient.Get(ctx, *podLookupKey, createdPod)).To(gomega.Succeed())
 							g.Expect(createdPod.Spec.SchedulingGates).Should(
-								gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+								gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 							)
 						}, util.Timeout, util.Interval).Should(gomega.Succeed())
 					})
@@ -672,7 +673,7 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 				pod1 := testingpod.MakePod("test-pod1", ns.Name).
 					Group("test-group").
 					GroupTotalCount("2").
-					Annotation(podcontroller.GroupFastAdmissionAnnotation, "true").
+					Annotation(podconstants.GroupFastAdmissionAnnotationKey, podconstants.GroupFastAdmissionAnnotationValue).
 					Queue("test-queue").
 					Obj()
 				pod1LookupKey := client.ObjectKeyFromObject(pod1)
@@ -714,7 +715,7 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 				pod2 := testingpod.MakePod("test-pod2", ns.Name).
 					Group("test-group").
 					GroupTotalCount("2").
-					Annotation(podcontroller.GroupFastAdmissionAnnotation, "true").
+					Annotation(podconstants.GroupFastAdmissionAnnotationKey, podconstants.GroupFastAdmissionAnnotationValue).
 					Queue("test-queue").
 					Obj()
 				pod2LookupKey := client.ObjectKeyFromObject(pod2)
@@ -1127,7 +1128,7 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 					Group("test-group").
 					GroupTotalCount("2").
 					Image("test-image", nil).
-					Annotation("kueue.x-k8s.io/retriable-in-group", "false").
+					Annotation(podconstants.RetriableInGroupAnnotationKey, podconstants.RetriableInGroupAnnotationValue).
 					Queue("test-queue").
 					Obj()
 				gomega.Expect(k8sClient.Create(ctx, replacementPod2)).Should(gomega.Succeed())
@@ -1477,14 +1478,14 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 				pod1 := testingpod.MakePod("test-pod1", ns.Name).
 					Group("test-group").
 					GroupTotalCount("2").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					Request(corev1.ResourceCPU, "1").
 					Queue("test-queue").
 					Obj()
 				pod2 := testingpod.MakePod("test-pod2", ns.Name).
 					Group("test-group").
 					GroupTotalCount("2").
-					PodGroupServingAnnotation(true).
+					PodGroupServingAnnotation().
 					Request(corev1.ResourceCPU, "1").
 					Queue("test-queue").
 					Obj()
@@ -1605,7 +1606,7 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 
 				wl := testing.MakeWorkload(workloadName, ns.Name).
 					Queue(lq.Name).
-					Annotation(podcontroller.IsGroupWorkloadAnnotationKey, podcontroller.IsGroupWorkloadAnnotationValue).
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					PodSets(
 						*testing.MakePodSet("leader", 1).PodSpec(pod1.Spec).Obj(),
 						*testing.MakePodSet("worker", 1).PodSpec(pod2.Spec).Obj(),
@@ -1891,7 +1892,7 @@ var _ = ginkgo.Describe("Pod controller interacting with scheduler", ginkgo.Orde
 					g.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}, createdPod)).
 						To(gomega.Succeed())
 					g.Expect(createdPod.Spec.SchedulingGates).Should(
-						gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+						gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 					)
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
@@ -2235,7 +2236,7 @@ var _ = ginkgo.Describe("Pod controller when TopologyAwareScheduling enabled", g
 		ginkgo.By("creating a pod which requires block", func() {
 			gomega.Expect(k8sClient.Create(ctx, pod)).Should(gomega.Succeed())
 			gomega.Expect(pod.Spec.SchedulingGates).Should(gomega.ContainElements(
-				corev1.PodSchedulingGate{Name: podcontroller.SchedulingGateName},
+				corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName},
 				corev1.PodSchedulingGate{Name: kueuealpha.TopologySchedulingGate},
 			))
 			gomega.Expect(pod.Labels).To(gomega.HaveKeyWithValue(kueuealpha.TASLabel, "true"))
@@ -2289,7 +2290,7 @@ var _ = ginkgo.Describe("Pod controller when TopologyAwareScheduling enabled", g
 			for _, p := range group {
 				gomega.Expect(k8sClient.Create(ctx, p)).To(gomega.Succeed())
 				gomega.Expect(p.Spec.SchedulingGates).To(gomega.ContainElements(
-					corev1.PodSchedulingGate{Name: podcontroller.SchedulingGateName},
+					corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName},
 					corev1.PodSchedulingGate{Name: kueuealpha.TopologySchedulingGate},
 				))
 				gomega.Expect(p.Labels).To(gomega.HaveKeyWithValue(kueuealpha.TASLabel, "true"))
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index a2e3a6517..18c20a584 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -63,36 +63,35 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 
 	ginkgo.When("Delete Topology", func() {
 		var (
-			topology *kueuealpha.Topology
+			tasFlavor *kueue.ResourceFlavor
+			topology  *kueuealpha.Topology
 		)
 
-		ginkgo.BeforeEach(func() {
-			topology = testing.MakeDefaultOneLevelTopology("topology")
-			gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
-		})
-
 		ginkgo.AfterEach(func() {
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
 			util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
 		})
 
-		ginkgo.It("should allow to delete topology", func() {
-			util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
-		})
+		ginkgo.When("ResourceFlavor does not exist", func() {
+			ginkgo.BeforeEach(func() {
+				topology = testing.MakeDefaultOneLevelTopology("topology")
+				gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
+			})
 
-		ginkgo.When("ResourceFlavor exist", func() {
-			var (
-				tasFlavor *kueue.ResourceFlavor
-			)
+			ginkgo.It("should allow to delete topology", func() {
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
+			})
+		})
 
+		ginkgo.When("ResourceFlavor exists", func() {
 			ginkgo.BeforeEach(func() {
 				tasFlavor = testing.MakeResourceFlavor("tas-flavor").
 					NodeLabel("node-group", "tas").
 					TopologyName(topology.Name).Obj()
 				gomega.Expect(k8sClient.Create(ctx, tasFlavor)).Should(gomega.Succeed())
-			})
 
-			ginkgo.AfterEach(func() {
-				util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
+				topology = testing.MakeDefaultOneLevelTopology("topology")
+				gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
 			})
 
 			ginkgo.It("should not allow to delete topology", func() {
@@ -106,9 +105,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 				})
 
 				ginkgo.By("delete topology", func() {
-					gomega.Eventually(func(g gomega.Gomega) {
-						g.Expect(k8sClient.Delete(ctx, topology)).Should(gomega.Succeed())
-					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+					gomega.Expect(k8sClient.Delete(ctx, topology)).Should(gomega.Succeed())
 				})
 
 				ginkgo.By("check topology still present", func() {
diff --git a/test/integration/singlecluster/webhook/jobs/pod_webhook_test.go b/test/integration/singlecluster/webhook/jobs/pod_webhook_test.go
index f4d4e1925..642e21671 100644
--- a/test/integration/singlecluster/webhook/jobs/pod_webhook_test.go
+++ b/test/integration/singlecluster/webhook/jobs/pod_webhook_test.go
@@ -26,7 +26,8 @@ import (
 
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/util/kubeversion"
 	testingpod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
 	"sigs.k8s.io/kueue/test/util"
@@ -56,7 +57,7 @@ var _ = ginkgo.Describe("Pod Webhook", func() {
 			gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
 			fwk.StartManager(ctx, cfg, managerSetup(
-				pod.SetupWebhook,
+				podcontroller.SetupWebhook,
 				jobframework.WithManageJobsWithoutQueueName(false),
 				jobframework.WithManagedJobsNamespaceSelector(mjnsSelector),
 				jobframework.WithKubeServerVersion(serverVersionFetcher),
@@ -98,7 +99,7 @@ var _ = ginkgo.Describe("Pod Webhook", func() {
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 				gomega.Expect(createdPod.Spec.SchedulingGates).To(
-					gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+					gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 					"Pod should have scheduling gate",
 				)
 
@@ -122,7 +123,7 @@ var _ = ginkgo.Describe("Pod Webhook", func() {
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 				gomega.Expect(createdPod.Spec.SchedulingGates).NotTo(
-					gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+					gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 					"Pod shouldn't have scheduling gate",
 				)
 
@@ -156,7 +157,7 @@ var _ = ginkgo.Describe("Pod Webhook", func() {
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 				gomega.Expect(createdPod.Spec.SchedulingGates).NotTo(
-					gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+					gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 					"Pod shouldn't have scheduling gate",
 				)
 
@@ -191,7 +192,7 @@ var _ = ginkgo.Describe("Pod Webhook", func() {
 			gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
 			fwk.StartManager(ctx, cfg, managerSetup(
-				pod.SetupWebhook,
+				podcontroller.SetupWebhook,
 				jobframework.WithManageJobsWithoutQueueName(true),
 				jobframework.WithManagedJobsNamespaceSelector(mjnsSelector),
 				jobframework.WithKubeServerVersion(serverVersionFetcher),
@@ -230,7 +231,7 @@ var _ = ginkgo.Describe("Pod Webhook", func() {
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 				gomega.Expect(createdPod.Spec.SchedulingGates).To(
-					gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+					gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 					"Pod should have scheduling gate",
 				)
 
@@ -254,7 +255,7 @@ var _ = ginkgo.Describe("Pod Webhook", func() {
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 				gomega.Expect(createdPod.Spec.SchedulingGates).NotTo(
-					gomega.ContainElement(corev1.PodSchedulingGate{Name: "kueue.x-k8s.io/admission"}),
+					gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}),
 					"Pod shouldn't have scheduling gate",
 				)
 
diff --git a/test/integration/singlecluster/webhook/jobs/statefulset_webhook_test.go b/test/integration/singlecluster/webhook/jobs/statefulset_webhook_test.go
index 7bf183024..5142bd722 100644
--- a/test/integration/singlecluster/webhook/jobs/statefulset_webhook_test.go
+++ b/test/integration/singlecluster/webhook/jobs/statefulset_webhook_test.go
@@ -27,7 +27,7 @@ import (
 
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/statefulset"
 	"sigs.k8s.io/kueue/pkg/util/kubeversion"
 	testingstatefulset "sigs.k8s.io/kueue/pkg/util/testingjobs/statefulset"
@@ -79,12 +79,12 @@ var _ = ginkgo.Describe("StatefulSet Webhook", func() {
 							gomega.Equal("user-queue"),
 							"Queue name should be injected to pod template labels",
 						)
-					g.Expect(createdStatefulSet.Spec.Template.Labels[pod.GroupNameLabel]).
+					g.Expect(createdStatefulSet.Spec.Template.Labels[podconstants.GroupNameLabel]).
 						To(
 							gomega.Equal(jobframework.GetWorkloadNameForOwnerWithGVK(createdStatefulSet.Name, "", appsv1.SchemeGroupVersion.WithKind("StatefulSet"))),
 							"Pod group name should be injected to pod template labels",
 						)
-					g.Expect(createdStatefulSet.Spec.Template.Annotations[pod.GroupTotalCountAnnotation]).
+					g.Expect(createdStatefulSet.Spec.Template.Annotations[podconstants.GroupTotalCountAnnotation]).
 						To(
 							gomega.Equal("1"),
 							"Pod group total count should be injected to pod template annotations",
diff --git a/test/util/e2e.go b/test/util/e2e.go
index 89d252fb0..4af8604fa 100644
--- a/test/util/e2e.go
+++ b/test/util/e2e.go
@@ -35,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	jobset "sigs.k8s.io/jobset/api/jobset/v1alpha2"
@@ -248,11 +249,14 @@ func RestartKueueController(ctx context.Context, k8sClient client.Client) {
 	rolloutOperatorDeployment(ctx, k8sClient, kcmKey)
 }
 
-func WaitForActivePodsAndTerminate(ctx context.Context, k8sClient client.Client, restClient *rest.RESTClient, cfg *rest.Config, namespace string, activePodsCount, exitCode int) {
+func WaitForActivePodsAndTerminate(ctx context.Context, k8sClient client.Client, restClient *rest.RESTClient, cfg *rest.Config, namespace string, activePodsCount, exitCode int, opts ...client.ListOption) {
 	var activePods []corev1.Pod
 	pods := corev1.PodList{}
+	podListOpts := &client.ListOptions{}
+	podListOpts.Namespace = namespace
+	podListOpts.ApplyOptions(opts)
 	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
-		g.Expect(k8sClient.List(ctx, &pods, client.InNamespace(namespace))).To(gomega.Succeed())
+		g.Expect(k8sClient.List(ctx, &pods, podListOpts)).To(gomega.Succeed())
 		activePods = make([]corev1.Pod, 0)
 		for _, p := range pods.Items {
 			if (len(p.Status.PodIP) != 0 && p.Status.PodIP != "0.0.0.0") && (p.Status.Phase == corev1.PodRunning || p.Status.Phase == corev1.PodPending) {
@@ -263,6 +267,7 @@ func WaitForActivePodsAndTerminate(ctx context.Context, k8sClient client.Client,
 	}, LongTimeout, Interval).Should(gomega.Succeed())
 
 	for _, p := range activePods {
+		klog.V(3).Info("Terminating pod", "pod", klog.KObj(&p))
 		cmd := []string{"/bin/sh", "-c", fmt.Sprintf("curl \"http://%s:8080/exit?code=%v&timeout=2s&wait=2s\"", p.Status.PodIP, exitCode)}
 		_, _, err := KExecute(ctx, cfg, restClient, namespace, p.Name, p.Spec.Containers[0].Name, cmd)
 		// TODO: remove the custom handling of 137 response once this is fixed in the agnhost image
diff --git a/test/util/util.go b/test/util/util.go
index c0451d3e5..da89131ac 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -62,7 +62,7 @@ import (
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
-	"sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/scheduler/preemption"
 	utiltas "sigs.k8s.io/kueue/pkg/util/tas"
@@ -187,7 +187,7 @@ func deleteAllPodsInNamespace(ctx context.Context, c client.Client, ns *corev1.N
 		g.Expect(client.IgnoreNotFound(c.List(ctx, &pods, client.InNamespace(ns.Name)))).
 			Should(gomega.Succeed(), "listing Pods with a finalizer in namespace %q", ns.Name)
 		for _, p := range pods.Items {
-			if controllerutil.RemoveFinalizer(&p, pod.PodFinalizer) {
+			if controllerutil.RemoveFinalizer(&p, podconstants.PodFinalizer) {
 				g.Expect(client.IgnoreNotFound(c.Update(ctx, &p))).Should(gomega.Succeed(), "removing finalizer")
 			}
 		}
@@ -762,7 +762,7 @@ func ExpectPodUnsuspendedWithNodeSelectors(ctx context.Context, k8sClient client
 	createdPod := &corev1.Pod{}
 	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
 		g.Expect(k8sClient.Get(ctx, key, createdPod)).To(gomega.Succeed())
-		g.Expect(createdPod.Spec.SchedulingGates).NotTo(gomega.ContainElement(corev1.PodSchedulingGate{Name: pod.SchedulingGateName}))
+		g.Expect(createdPod.Spec.SchedulingGates).NotTo(gomega.ContainElement(corev1.PodSchedulingGate{Name: podconstants.SchedulingGateName}))
 		g.Expect(createdPod.Spec.NodeSelector).To(gomega.BeComparableTo(ns))
 	}, Timeout, Interval).Should(gomega.Succeed())
 }