Repository: kubernetes-sigs/kueue

Issue:
Infinite preemption loop in fair sharing
**What happened**:

An infinite preemption loop occurs in a hierarchical cohort scenario with Kueue's Pod integration.
When a higher-priority workload preempts a lower-priority one from a different queue, the Deployment controller recreates the preempted workload, causing it to be readmitted. 
Once readmitted, the preemption occurs again, triggering this infinite loop.

I'll show an edge case where this infinite preemption behavior occurs. Not all preemption operations using hierarchical cohorts behave in this way.

Cohort Structure
--- Cohort Root (NM: 0m)
-------- Cohort Guaranteed (NM: 300m)
------------- Clusterqueue Guaranteed (NM: 0m)
-------- ClusterQueue BestEffortTenantA (NM: 0m)
--------  ClusterQueue BestEffortTenantB (NM: 0m)

Event Sequence Leading to the Issue
   1. BestEffortTenantA1 (100m) admitted
   2. BestEffortTenantA2 (100m) admitted
   3. BestEffortTenantB1 (50m) admitted
   4. BestEffortTenantB2 (50m) admitted
   5. Guaranteed1 (50m) created → preempts BestEffortTenantA2
   6. Guaranteed1 (50m) admitted
   7. Guaranteed2 (100m) created → triggers infinite loop:
      - BestEffortTenantB2 preempted
      - Guaranteed2 remains SchedulingGated
      - BestEffortTenantB2 recreated by Deployment
      - BestEffortTenantB2 admitted
      - Loop repeats
   8. After a few minutes, BestEffortTenantB2 resumes running without being preempted, while Guaranteed2 remains in the SchedulingGated state.
      
 -------------------------------------------------------
  
Pods status

```
> kubectl get pods
NAME                                        READY   STATUS            RESTARTS   AGE
best-effort-tenant-a-1-5576db6f68-nms98     1/1     Running           0          55s
best-effort-tenant-a-2-695bd7b4c-b5n4l      0/1     SchedulingGated   0          43s
best-effort-tenant-b-1-68596f9f8b-52p5z     1/1     Running           0          50s
best-effort-tenant-b-2-69dc4f5797-l4vq6     0/1     Terminating       0          5s
best-effort-tenant-b-2-69dc4f5797-l4vq6     0/1     SchedulingGated   0          2s
guaranteed-1-7c96f8db5-n6575                1/1     Running           0          43s
guaranteed-2-79f59d8b74-7ft4b               0/1     SchedulingGated   0          39s

```

**What you expected to happen**:

Preemption should complete successfully without entering an infinite loop.

**How to reproduce it (as minimally and precisely as possible)**:

1. Apply core configuration (cohorts, queues, and priority classes):
```
---
# Root Cohort
apiVersion: kueue.x-k8s.io/v1alpha1
kind: Cohort
metadata:
  name: root
---
# Guaranteed Cohort
apiVersion: kueue.x-k8s.io/v1alpha1
kind: Cohort
metadata:
  name: guaranteed
spec:
  parent: root
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default"
      resources:
      - name: "cpu"
        nominalQuota: "300m"
---
# Guaranteed ClusterQueue
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "guaranteed"
spec:
  namespaceSelector: {}
  preemption:
    reclaimWithinCohort: LowerPriority
    borrowWithinCohort:
      policy: LowerPriority
      maxPriorityThreshold: 100
    withinClusterQueue: Never
  cohort: guaranteed
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default"
      resources:
      - name: "cpu"
        nominalQuota: "0m"
---
# Best-effort ClusterQueues
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "best-effort-tenant-a"
spec:
  namespaceSelector: {}
  preemption:
    reclaimWithinCohort: Never
    borrowWithinCohort:
      policy: Never
    withinClusterQueue: Never
  cohort: root
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default"
      resources:
      - name: "cpu"
        nominalQuota: "0m"
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "best-effort-tenant-b"
spec:
  namespaceSelector: {}
  preemption:
    reclaimWithinCohort: Never
    borrowWithinCohort:
      policy: Never
    withinClusterQueue: Never
  cohort: root
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default"
      resources:
      - name: "cpu"
        nominalQuota: "0m"
---
# LocalQueues
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  name: guaranteed
spec:
  clusterQueue: guaranteed
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  name: best-effort-tenant-a
spec:
  clusterQueue: best-effort-tenant-a
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  name: best-effort-tenant-b
spec:
  clusterQueue: best-effort-tenant-b
---
# Priority Classes
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: low
value: 70
globalDefault: false
---
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: medium
value: 102
globalDefault: false
---
# Resource Flavor
apiVersion: kueue.x-k8s.io/v1beta1
kind: ResourceFlavor
metadata:
  name: default
---
```
2. Create best-effort workloads:
```
# Workloads - Best Effort A
apiVersion: apps/v1
kind: Deployment
metadata:
  name: best-effort-tenant-a-1
  labels: {app: best-effort-tenant-a-1, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-a}
spec:
  replicas: 1
  selector:
    matchLabels: {app: best-effort-tenant-a-1}
  template:
    metadata:
      labels: {app: best-effort-tenant-a-1, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-a}
    spec:
      priorityClassName: low
      terminationGracePeriodSeconds: 1
      containers:
      - name: main
        image: registry.k8s.io/pause:3.9
        resources:
          requests:
            cpu: "100m"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: best-effort-tenant-a-2
  labels: {app: best-effort-tenant-a-2, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-a}
spec:
  replicas: 1
  selector:
    matchLabels: {app: best-effort-tenant-a-2}
  template:
    metadata:
      labels: {app: best-effort-tenant-a-2, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-a}
    spec:
      priorityClassName: low
      terminationGracePeriodSeconds: 1
      containers:
      - name: main
        image: registry.k8s.io/pause:3.9
        resources:
          requests:
            cpu: "100m"
---
# Workloads - Best Effort B
apiVersion: apps/v1
kind: Deployment
metadata:
  name: best-effort-tenant-b-1
  labels: {app: best-effort-tenant-b-1, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-b}
spec:
  replicas: 1
  selector:
    matchLabels: {app: best-effort-tenant-b-1}
  template:
    metadata:
      labels: {app: best-effort-tenant-b-1, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-b}
    spec:
      priorityClassName: low
      terminationGracePeriodSeconds: 1
      containers:
      - name: main
        image: registry.k8s.io/pause:3.9
        resources:
          requests:
            cpu: "50m"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: best-effort-tenant-b-2
  labels: {app: best-effort-tenant-b-2, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-b}
spec:
  replicas: 1
  selector:
    matchLabels: {app: best-effort-tenant-b-2}
  template:
    metadata:
      labels: {app: best-effort-tenant-b-2, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort-tenant-b}
    spec:
      priorityClassName: low
      terminationGracePeriodSeconds: 1
      containers:
      - name: main
        image: registry.k8s.io/pause:3.9
        resources:
          requests:
            cpu: "50m"
 ```
 3. Create first guaranteed workload (triggers normal preemption):
 ```
apiVersion: apps/v1
kind: Deployment
metadata:
  name: guaranteed-1
  labels: {app: guaranteed-1, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed}
spec:
  replicas: 1
  selector:
    matchLabels: {app: guaranteed-1}
  template:
    metadata:
      labels: {app: guaranteed-1, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed}
    spec:
      priorityClassName: medium
      terminationGracePeriodSeconds: 1
      containers:
      - name: main
        image: registry.k8s.io/pause:3.9
        resources:
          requests:
            cpu: "50m"
```
4. Create second guaranteed workload (triggers infinite loop):
```
apiVersion: apps/v1
kind: Deployment
metadata:
  name: guaranteed-2
  labels: {app: guaranteed-2, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed}
spec:
  replicas: 1
  selector:
    matchLabels: {app: guaranteed-2}
  template:
    metadata:
      labels: {app: guaranteed-2, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed}
    spec:
      priorityClassName: medium
      terminationGracePeriodSeconds: 1
      containers:
      - name: main
        image: registry.k8s.io/pause:3.9
        resources:
          requests:
            cpu: "100m"
```

**Anything else we need to know?**:

kueue-manager logs
```
{"level":"Level(-2)","ts":"2024-12-09T21:21:39.997044887Z","logger":"workload-reconciler","caller":"queue/manager.go:528","msg":"Attempting to move workloads","workload":{"name":"pod-best-effort-tenant-b-2-69dc4f5797-gbndl-be03a","namespace":"default"},"queue":"best-effort-tenant-b","status":"admitted","cohort":"root","root":"root"}
{"level":"Level(-2)","ts":"2024-12-09T21:21:39.997158095Z","logger":"scheduler","caller":"scheduler/scheduler.go:543","msg":"Workload assumed in the cache","attemptCount":115,"workload":{"name":"pod-best-effort-tenant-b-2-69dc4f5797-2jpd6-289ea","namespace":"default"},"clusterQueue":{"name":"best-effort-tenant-b"}}
{"level":"info","ts":"2024-12-09T21:21:39.997385095Z","logger":"scheduler","caller":"scheduler/scheduler.go:365","msg":"Workload skipped from admission because it's already assumed or admitted","attemptCount":116,"workload":{"name":"pod-best-effort-tenant-b-2-69dc4f5797-2jpd6-289ea","namespace":"default"},"clusterQueue":{"name":"best-effort-tenant-b"},"workload":{"name":"pod-best-effort-tenant-b-2-69dc4f5797-2jpd6-289ea","namespace":"default"}}
{"level":"error","ts":"2024-12-09T21:21:40.00190772Z","logger":"scheduler","caller":"scheduler/scheduler.go:263","msg":"Failed to preempt workloads","attemptCount":116,"workload":{"name":"pod-guranateed-2-788b9dd5bb-zbvmc-667aa","namespace":"default"},"clusterQueue":{"name":"guranateed"},"error":"Operation cannot be fulfilled on workloads.kueue.x-k8s.io \"pod-best-effort-tenant-b-2-69dc4f5797-2jpd6-289ea\": the object has been modified; please apply your changes to the latest version and try again","stacktrace":"sigs.k8s.io/kueue/pkg/scheduler.(*Scheduler).schedule\n\t/workspace/pkg/scheduler/scheduler.go:263\nsigs.k8s.io/kueue/pkg/util/wait.untilWithBackoff.func1\n\t/workspace/pkg/util/wait/backoff.go:43\nk8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1\n\t/workspace/vendor/k8s.io/apimachinery/pkg/util/wait/backoff.go:226\nk8s.io/apimachinery/pkg/util/wait.BackoffUntil\n\t/workspace/vendor/k8s.io/apimachinery/pkg/util/wait/backoff.go:227\nsigs.k8s.io/kueue/pkg/util/wait.untilWithBackoff\n\t/workspace/pkg/util/wait/backoff.go:42\nsigs.k8s.io/kueue/pkg/util/wait.UntilWithBackoff\n\t/workspace/pkg/util/wait/backoff.go:34"}
{"level":"Level(-2)","ts":"2024-12-09T21:21:40.002000345Z","logger":"scheduler","caller":"scheduler/scheduler.go:645","msg":"Workload re-queued","attemptCount":116,"workload":{"name":"pod-best-effort-tenant-a-1-5576db6f68-hjsv2-9fc95","namespace":"default"},"clusterQueue":{"name":"best-effort-tenant-a"},"queue":{"name":"best-effort-tenant-a","namespace":"default"},"requeueReason":"","added":true,"status":""}
{"level":"Level(-2)","ts":"2024-12-09T21:21:40.006699845Z","logger":"scheduler","caller":"scheduler/scheduler.go:567","msg":"Workload successfully admitted and assigned flavors","attemptCount":115,"workload":{"name":"pod-best-effort-tenant-b-2-69dc4f5797-2jpd6-289ea","namespace":"default"},"clusterQueue":{"name":"best-effort-tenant-b"},"assignments":[{"name":"main","flavors":{"cpu":"default"},"resourceUsage":{"cpu":"50m"},"count":1}]}
{"level":"debug","ts":"2024-12-09T21:21:40.006734262Z","logger":"events","caller":"recorder/recorder.go:104","msg":"Quota reserved in ClusterQueue best-effort-tenant-b, wait time since queued was 2s","type":"Normal","object":{"kind":"Workload","namespace":"default","name":"pod-best-effort-tenant-b-2-69dc4f5797-2jpd6-289ea","uid":"df8be6f5-0bba-4bd4-9f00-39b3209aa82c","apiVersion":"kueue.x-k8s.io/v1beta1","resourceVersion":"556812"},"reason":"QuotaReserved"}
{"level":"debug","ts":"2024-12-09T21:21:40.006753928Z","logger":"events","caller":"recorder/recorder.go:104","msg":"Admitted by ClusterQueue best-effort-tenant-b, wait time since reservation was 0s","type":"Normal","object":{"kind":"Workload","namespace":"default","name":"pod-best-effort-tenant-b-2-69dc4f5797-2jpd6-289ea","uid":"df8be6f5-0bba-4bd4-9f00-39b3209aa82c","apiVersion":"kueue.x-k8s.io/v1beta1","resourceVersion":"556812"},"reason":"Admitted"}
```
**Environment**:
- Kubernetes version: v1.31.0
- Kueue version: v0.10.0-rc.3-20-g6df4a225-dirty
- Cloud provider or hardware configuration: local setup ( minikube ) & gke
- OS (e.g: `cat /etc/os-release`):
- Kernel (e.g. `uname -a`):
- Install tools:
- Others:

Hints from issue comments:
@mimowo: @yuvalaz99 thank you for reporting the issue with the detailed scenario. Let me ask some questions to triage it. I could probably look into this closer by EOW, or next week.

There are a couple of features used here: HierarchicalCohorts, Deployments, borrowWithinCohort. Do you have some evidence for which one is the actual culprit? Do you have a temporary workaround for the issue? Would you like to work on the fix? @yuvalaz99: > @yuvalaz99 thank you for reporting the issue with the detailed scenario. Let me ask some questions to triage it. I could probably look into this closer by EOW, or next week.
> 
> There are a couple of features used here: HierarchicalCohorts, Deployments, borrowWithinCohort. Do you have some evidence for which one is the actual culprit? Do you have a temporary workaround for the issue? Would you like to work on the fix?

Thank you for the quick response :)
So iv'e verified that with the same setup and without HierarchicalCohorts this issue does not occur.

Cohort Structure
--- Cohort Root (NM: 0m)
-------- Clusterqueue Guaranteed (NM: 300m)
-------- ClusterQueue BestEffortTenantA (NM: 0m)
-------- ClusterQueue BestEffortTenantB (NM: 0m)
@mimowo: I see, thank you for the update, it feels the next step will be to replicate, preferably in integration tests, and debug where is the culprit.
@mbobrovskyi: /assign @vladikkuzn @gabesaba: cc @gabesaba @gabesaba: Hi @yuvalaz99, I was unable to reproduce the issue. Were there any additional options/feature gates you configured when running Kueue? Are you still able to reproduce this issue using Kueue 0.10.0?

**Setup:**
I ran Kueue 0.10.0 on GKE 1.31, enabling the pod and deployment integrations.

**Steps:**

I applied the configurations you provided, in order. I observed different preemptions: best-effort-tenant-b-2 when guaranteed-1 was created, and best-effort-tenant-b-1 when guaranteed-2 was created. guaranteed-1 and guaranteed-2 admitted and scheduled successfully, with best-effort-tenant-b-1 and best-effort-tenant-b-2 remaining gated post preemption.

To match your description more closely, I also created the best effort workloads sequentially (since we [sort preemption candidates based on admission timestamp](https://github.com/kubernetes-sigs/kueue/blob/bf4657adc4e82b1458a8e0f78e0e21af15486d7b/pkg/scheduler/preemption/preemption.go#L587)) [a1, b1, b2, a2], to reproduce the preemptions of a2 and b2 you mentioned. In this case, guaranteed-1 and guaranteed-2 were still able to admit and schedule.@yuvalaz99: Hi @gabesaba,
After some further investigation, I believe the issue is broader than just hierarchical cohorts.

The core problem appears to be that when the preempting ClusterQueue has a higher weighted share than the target ClusterQueue. The workload that was preempted gets admitted again to maintain fairness in the system ( which will be again preempted ).

I’ll provide a minimal configuration to reproduce this issue.

Cohort Structure
--- Cohort Root (NM: 0m)
-------- Clusterqueue GuaranteedTenantA (NM: 150m)
-------- Clusterqueue GuaranteedTenantB (NM: 150m)
-------- ClusterQueue BestEffort (NM: 0m)

**Event Sequence Leading to the Issue**

GuaranteedTenantA1 (250m) admitted
BestEffort1 (50m) admitted
GuaranteedTenantA2 (50m) created → triggers infinite loop:
- BestEffort1 preempted
- BestEffort1 recreated by its Deployment
- BestEffort1 & GuaranteedTenantA2 remain in SchedulingGated
- BestEffort1 admitted again **( because ClusterQueue BestEffort has lower Wighted Share value than ClusterQueue GuaranteedTenantA )**

**How to reproduce the issue:**

1. Apply core configuration (cohorts, queues, resourceflavors and priority classes):
```
---
# Root Cohort
apiVersion: kueue.x-k8s.io/v1alpha1
kind: Cohort
metadata:
  name: testing
---
# Guaranteed ClusterQueue
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "guaranteed-tenant-a"
spec:
  namespaceSelector: {}
  preemption:
    reclaimWithinCohort: LowerPriority
    borrowWithinCohort:
      policy: LowerPriority
      maxPriorityThreshold: 100
    withinClusterQueue: Never
  cohort: testing
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default"
      resources:
      - name: "cpu"
        nominalQuota: "150m"
---
# Guaranteed ClusterQueue
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "guaranteed-tenant-b"
spec:
  namespaceSelector: {}
  preemption:
    reclaimWithinCohort: LowerPriority
    borrowWithinCohort:
      policy: LowerPriority
      maxPriorityThreshold: 100
    withinClusterQueue: Never
  cohort: testing
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default"
      resources:
      - name: "cpu"
        nominalQuota: "150m"        
---
# Best-effort ClusterQueues
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "best-effort"
spec:
  namespaceSelector: {}
  preemption:
    reclaimWithinCohort: Never
    borrowWithinCohort:
      policy: Never
    withinClusterQueue: Never
  cohort: testing
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default"
      resources:
      - name: "cpu"
        nominalQuota: "0m"
---
# LocalQueues
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  name: guaranteed-tenant-a
spec:
  clusterQueue: guaranteed-tenant-a
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  name: guaranteed-tenant-b
spec:
  clusterQueue: guaranteed-tenant-b
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  name: best-effort
spec:
  clusterQueue: best-effort
---
# Priority Classes
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: low
value: 70
globalDefault: false
---
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: medium
value: 102
globalDefault: false
---
# Resource Flavor
apiVersion: kueue.x-k8s.io/v1beta1
kind: ResourceFlavor
metadata:
  name: default
```

2. Create workloads that will utilize the available quota
```
---
# Workloads - Guaranteed Tenant A
apiVersion: apps/v1
kind: Deployment
metadata:
 name: guaranteed-tenant-a-1
 labels: {app: guaranteed-tenant-a-1, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed-tenant-a}
spec:
 replicas: 1
 selector:
   matchLabels: {app: guaranteed-tenant-a-1}
 template:
   metadata:
     labels: {app: guaranteed-tenant-a-1, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed-tenant-a}
   spec:
     priorityClassName: medium
     terminationGracePeriodSeconds: 1
     containers:
     - name: main
       image: registry.k8s.io/pause:3.9
       resources:
         requests:
           cpu: "250m"
---
# Workloads - Best Effort 
apiVersion: apps/v1
kind: Deployment
metadata:
  name: best-effort-1
  labels: {app: best-effort-1, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort}
spec:
  replicas: 1
  selector:
    matchLabels: {app: best-effort-1}
  template:
    metadata:
      labels: {app: best-effort-1, kueue-job: "true", kueue.x-k8s.io/queue-name: best-effort}
    spec:
      priorityClassName: low
      terminationGracePeriodSeconds: 1
      containers:
      - name: main
        image: registry.k8s.io/pause:3.9
        resources:
          requests:
            cpu: "50m"
```

3. Create workload that will trigger infinite preemption loop
```
---
# Workloads - Guaranteed Tenant A
apiVersion: apps/v1
kind: Deployment
metadata:
 name: guaranteed-tenant-a-2
 labels: {app: guaranteed-tenant-a-2, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed-tenant-a}
spec:
 replicas: 1
 selector:
   matchLabels: {app: guaranteed-tenant-a-2}
 template:
   metadata:
     labels: {app: guaranteed-tenant-a-2, kueue-job: "true", kueue.x-k8s.io/queue-name: guaranteed-tenant-a}
   spec:
     priorityClassName: medium
     terminationGracePeriodSeconds: 1
     containers:
     - name: main
       image: registry.k8s.io/pause:3.9
       resources:
         requests:
           cpu: "50m"
```

**Expected pods status**
```
> kubectl get pods
NAME                                     READY   STATUS            RESTARTS   AGE
best-effort-1-6467566f99-9mcxj           0/1     Terminating       0          3s
best-effort-1-6467566f99-bglnp           0/1     SchedulingGated   0          0s
guaranteed-tenant-a-1-78bf7b9448-ps98v   1/1     Running           0          76s
guaranteed-tenant-a-2-d9cf8dc5-fxx69     0/1     SchedulingGated   0          65s
```

**Our Kueue configuration:**
```
  managerConfig:
    controllerManagerConfigYaml: |-
      apiVersion: config.kueue.x-k8s.io/v1beta1
      kind: Configuration
      metrics:
        enableClusterQueueResources: true
      waitForPodsReady:
       enable: true
       timeout: 168h
       blockAdmission: false
      integrations:
        frameworks:
        - "pod"
        podOptions:
          namespaceSelector:
            matchExpressions:
            - key: kubernetes.io/metadata.name
              operator: In
              values:
              - default
          podSelector:
            matchExpressions:
            - key: kueue.x-k8s.io/queue-name
              operator: Exists
      fairSharing:
        enable: true
        preemptionStrategies: [LessThanInitialShare]
      resources:
        excludeResourcePrefixes: [
          ephemeral-storage,
          memory,
          networking.gke.io.networks
        ]
```@gabesaba: Thanks for the additional info, @yuvalaz99. I'll take a look this week into this more generalized issue you described.

I want to note: in the current state, Hierarchical Cohorts and Fair Sharing are incompatible - their usage together results in undefined behavior. We are working on making Fair Sharing and Hierarchical Cohorts compatible now. That may be a culprit in the initial issue you described, but not the latest, as the cohort structure in your latest issue description is flat.@yuvalaz99: @gabesaba Thank you for your response and for looking into this!

Please note that the new information shows this unexpected behavior also occurs when hierarchical cohorts are not used.

If you need any more information, I’ll be happy to help :)@gabesaba: > [@gabesaba](https://github.com/gabesaba) Thank you for your response and for looking into this!
> 
> Please note that the new information shows this unexpected behavior also occurs when hierarchical cohorts are not used.
> 
> If you need any more information, I’ll be happy to help :)

I was able to reproduce the issue - though it didn't result in an infinite preemption loop, but some unnecessary preemptions (`best-effort-1` workload was preempted a few times, before it was left alone by `guaranteed-tenant-a-2`). I'll provide an estimate next week of by when we can get this fixed.@gabesaba: We're reproducing the bug in an integration test (#4030). Afterwords, we'll implement a fix.@gabesaba: > After some further investigation, I believe the issue is broader than just hierarchical cohorts.

@yuvalaz99, could you update the issue title please, to more accurately reflect this? Something like "Preemption Loop in Fair Sharing"@yuvalaz99: > We're reproducing the bug in an integration test ([#4030](https://github.com/kubernetes-sigs/kueue/pull/4030)). Afterwords, we'll implement a fix.

I'm glad you were able to reproduce it. Thanks :)@gabesaba: The root cause is the below-threshold logic:
https://github.com/kubernetes-sigs/kueue/blob/d02a7649e63fba77d546adaa8be11c72624009f5/pkg/scheduler/preemption/preemption.go#L378-L381

When doing fair preemptions, even though best effort queue + workload has a lower DominantResourceShare than guaranteed queue + workload, this logic is short circuited by belowThreshold, and causing Kueue to preempt anyway. Then, when we get to scheduling, best effort Kueue has a lower DominantResourceShare than guaranteed-1, so it is scheduled. Then the loop occurs.

I will think about how to fix this. In the mean time, can you try removing `spec.preemption.borrowWithinCohort` from the guaranteed queues? This should resolve the problem - and hopefully be compatible with your setup otherwise

@yuvalaz99: > The root cause is the below-threshold logic:
> 
> [kueue/pkg/scheduler/preemption/preemption.go](https://github.com/kubernetes-sigs/kueue/blob/d02a7649e63fba77d546adaa8be11c72624009f5/pkg/scheduler/preemption/preemption.go#L378-L381)
> 
> Lines 378 to 381 in [d02a764](/kubernetes-sigs/kueue/commit/d02a7649e63fba77d546adaa8be11c72624009f5)
> 
>  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 { 
> When doing fair preemptions, even though best effort queue + workload has a lower DominantResourceShare than guaranteed queue + workload, this logic is short circuited by belowThreshold, and causing Kueue to preempt anyway. Then, when we get to scheduling, best effort Kueue has a lower DominantResourceShare than guaranteed-1, so it is scheduled. Then the loop occurs.
> 
> I will think about how to fix this. In the mean time, can you try removing `spec.preemption.borrowWithinCohort` from the guaranteed queues? This should resolve the problem - and hopefully be compatible with your setup otherwise

@gabesaba Thanks for your reply :) 

Currently, we are required to use this parameter to meet our business requirements.
Until Kueue introduces either:
- Fair sharing between LQ’s
- CQ preemption protection

we won’t be able to remove this parameter.

@gabesaba: /assign@gabesaba: @yuvalaz99, please take a look at #4165, and see if you think it resolves your issue. Please note that this change will make `spec.preemption.borrowWithinCohort` not have any effect when `FairSharing` is enabled. Furthermore, if this PR merges, we will likely make this parameter combination invalid, if we don't find a compelling and understandable/intuitive way to make these features work together.

Was this preemption below priority, while ignoring fair sharing value, something that you were relying on for your business requirements?

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/manager/manager.yaml b/charts/kueue/templates/manager/manager.yaml
index c23b031d5..b1be101ff 100644
--- a/charts/kueue/templates/manager/manager.yaml
+++ b/charts/kueue/templates/manager/manager.yaml
@@ -47,6 +47,9 @@ spec:
         - containerPort: 9443
           name: webhook-server
           protocol: TCP
+        - containerPort: 8443
+          name: metrics
+          protocol: TCP
         readinessProbe:
           httpGet:
             path: /readyz
diff --git a/charts/kueue/values.yaml b/charts/kueue/values.yaml
index d51535012..5418db198 100644
--- a/charts/kueue/values.yaml
+++ b/charts/kueue/values.yaml
@@ -134,10 +134,10 @@ managerConfig:
 # ports definition for metricsService and webhookService.
 metricsService:
   ports:
-    - name: https
+    - name: metrics
       port: 8443
       protocol: TCP
-      targetPort: https
+      targetPort: metrics
   type: ClusterIP
   annotations: {}
 webhookService:
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/default/kustomization.yaml b/config/default/kustomization.yaml
index 39d0791c3..c547ad1d3 100644
--- a/config/default/kustomization.yaml
+++ b/config/default/kustomization.yaml
@@ -49,6 +49,9 @@ patches:
 # Expose port used by the visibility server
 - path: manager_visibility_patch.yaml
 
+# Expose port used by the metrics service
+- path: manager_metrics_patch.yaml
+
 # [CERTMANAGER] To enable cert-manager, uncomment all sections with 'CERTMANAGER'.
 # Uncomment 'CERTMANAGER' sections in crd/kustomization.yaml to enable the CA injection in the admission webhooks.
 # 'CERTMANAGER' needs to be enabled to use ca injection
diff --git a/config/default/manager_metrics_patch.yaml b/config/default/manager_metrics_patch.yaml
new file mode 100644
index 000000000..f46769c00
--- /dev/null
+++ b/config/default/manager_metrics_patch.yaml
@@ -0,0 +1,15 @@
+# This patch exposes 8443 port used by metrics service
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: controller-manager
+  namespace: system
+spec:
+  template:
+    spec:
+      containers:
+        - name: manager
+          ports:
+          - containerPort: 8443
+            name: metrics
+            protocol: TCP
diff --git a/pkg/cache/clusterqueue.go b/pkg/cache/clusterqueue.go
index e5a5a303e..79f48ba40 100644
--- a/pkg/cache/clusterqueue.go
+++ b/pkg/cache/clusterqueue.go
@@ -274,10 +274,6 @@ func (c *clusterQueue) inactiveReason() (string, string) {
 		}
 
 		if features.Enabled(features.TopologyAwareScheduling) && len(c.tasFlavors) > 0 {
-			if c.HasParent() {
-				reasons = append(reasons, kueue.ClusterQueueActiveReasonNotSupportedWithTopologyAwareScheduling)
-				messages = append(messages, "TAS is not supported for cohorts")
-			}
 			if len(c.multiKueueAdmissionChecks) > 0 {
 				reasons = append(reasons, kueue.ClusterQueueActiveReasonNotSupportedWithTopologyAwareScheduling)
 				messages = append(messages, "TAS is not supported with MultiKueue admission check")
@@ -312,9 +308,7 @@ func (c *clusterQueue) isTASViolated() bool {
 			return true
 		}
 	}
-	return c.HasParent() ||
-		len(c.multiKueueAdmissionChecks) > 0 ||
-		len(c.provisioningAdmissionChecks) > 0
+	return len(c.multiKueueAdmissionChecks) > 0 || len(c.provisioningAdmissionChecks) > 0
 }
 
 // UpdateWithFlavors updates a ClusterQueue based on the passed ResourceFlavors set.
@@ -514,7 +508,7 @@ func (c *clusterQueue) updateWorkloadUsage(wi *workload.Info, m int64) {
 			removeUsage(c, fr, q)
 		}
 	}
-	if features.Enabled(features.TopologyAwareScheduling) && wi.IsUsingTAS() {
+	if features.Enabled(features.TopologyAwareScheduling) && wi.IsRequestingTAS() {
 		for tasFlavor, tasUsage := range wi.TASUsage() {
 			if tasFlvCache := c.tasFlavorCache(tasFlavor); tasFlvCache != nil {
 				if m == 1 {
diff --git a/pkg/cache/clusterqueue_snapshot.go b/pkg/cache/clusterqueue_snapshot.go
index ce19a4b35..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)
 	}
@@ -119,6 +119,14 @@ func (c *ClusterQueueSnapshot) Fits(usage workload.Usage) bool {
 			return false
 		}
 	}
+	for tasFlavor, flvUsage := range usage.TAS {
+		// We assume the `tasFlavor` is already in the snapshot as this was
+		// already checked earlier during flavor assignment, and the set of
+		// flavors is immutable in snapshot.
+		if !c.TASFlavors[tasFlavor].Fits(flvUsage) {
+			return false
+		}
+	}
 	return true
 }
 
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/cache/tas_flavor_snapshot.go b/pkg/cache/tas_flavor_snapshot.go
index a04e97cde..f361efb08 100644
--- a/pkg/cache/tas_flavor_snapshot.go
+++ b/pkg/cache/tas_flavor_snapshot.go
@@ -31,6 +31,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/resources"
 	utiltas "sigs.k8s.io/kueue/pkg/util/tas"
+	"sigs.k8s.io/kueue/pkg/workload"
 )
 
 var (
@@ -282,6 +283,23 @@ type tasPodSetAssignmentResult struct {
 
 type FlavorTASRequests []TASPodSetRequests
 
+// Fits checks if the snapshot has enough capacity to accommodate the workload
+func (s *TASFlavorSnapshot) Fits(flavorUsage workload.TASFlavorUsage) bool {
+	for _, domainUsage := range flavorUsage {
+		domainID := utiltas.DomainID(domainUsage.Values)
+		leaf, found := s.leaves[domainID]
+		if !found {
+			return false
+		}
+		remainingCapacity := leaf.freeCapacity.Clone()
+		remainingCapacity.Sub(leaf.tasUsage)
+		if domainUsage.SinglePodRequests.CountIn(remainingCapacity) < domainUsage.Count {
+			return false
+		}
+	}
+	return true
+}
+
 // FindTopologyAssignmentsForFlavor returns TAS assignment, if possible, for all
 // the TAS requests in the flavor handled by the snapshot.
 // The simulateEmpty parameter allows to look for the assignment under the
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 4b8c8788d..6ab1cc114 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner.go
@@ -47,12 +47,44 @@ 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
 }
 
+// UpdateForTASResult updates the Assignment with the TAS result
+func (assignment *Assignment) UpdateForTASResult(result cache.TASAssignmentsResult) {
+	for psName, psResult := range result {
+		psAssignment := assignment.podSetAssignmentByName(psName)
+		psAssignment.TopologyAssignment = psResult.TopologyAssignment
+	}
+	assignment.Usage.TAS = assignment.computeTASUsage()
+}
+
+// TASUsage computes the TAS usage for the assignment
+func (assignment *Assignment) computeTASUsage() workload.TASUsage {
+	result := make(workload.TASUsage)
+	for _, psa := range assignment.PodSets {
+		if psa.TopologyAssignment != nil {
+			singlePodRequests := resources.NewRequests(psa.Requests).ScaledDown(int64(psa.Count))
+			for _, flv := range psa.Flavors {
+				if _, ok := result[flv.Name]; !ok {
+					result[flv.Name] = make(workload.TASFlavorUsage, 0)
+				}
+				for _, domain := range psa.TopologyAssignment.Domains {
+					result[flv.Name] = append(result[flv.Name], workload.TopologyDomainRequests{
+						Values:            domain.Values,
+						SinglePodRequests: singlePodRequests.Clone(),
+						Count:             domain.Count,
+					})
+				}
+			}
+		}
+	}
+	return result
+}
+
 // Borrows return whether assignment requires borrowing.
 func (a *Assignment) Borrows() bool {
 	return a.Borrowing
@@ -358,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,
@@ -383,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
@@ -414,10 +448,7 @@ func (a *FlavorAssigner) assignFlavors(log logr.Logger, counts []int32) Assignme
 				assignment.representativeMode = ptr.To(Preempt)
 			} else {
 				// All PodSets fit, we just update the TopologyAssignments
-				for psName, psResult := range result {
-					psAssignment := assignment.podSetAssignmentByName(psName)
-					psAssignment.TopologyAssignment = psResult.TopologyAssignment
-				}
+				assignment.UpdateForTASResult(result)
 			}
 		}
 		if assignment.RepresentativeMode() == Preempt {
@@ -454,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 7a530ee68..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,16 +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}
+func (e *entry) assignmentUsage() workload.Usage {
+	return e.assignment.Usage
 }
 
 // nominate returns the workloads with their requirements (resource flavors, borrowing) if
@@ -332,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
@@ -342,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 {
@@ -357,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)
 	}
@@ -380,15 +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.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 {
@@ -409,6 +413,13 @@ 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)
+	return assignment, targets
+}
+
+func (s *Scheduler) getInitialAssignments(log logr.Logger, wl *workload.Info, snap *cache.Snapshot) (flavorassigner.Assignment, []*preemption.Target) {
 	cq := snap.ClusterQueue(wl.ClusterQueue)
 	flvAssigner := flavorassigner.New(wl, cq, snap.ResourceFlavors, s.fairSharing.Enable, preemption.NewOracle(s.preemptor, snap))
 	fullAssignment := flvAssigner.Assign(log, nil)
@@ -448,6 +459,31 @@ func (s *Scheduler) getAssignments(log logr.Logger, wl *workload.Info, snap *cac
 	return fullAssignment, nil
 }
 
+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
+		if len(targets) > 0 {
+			var targetWorkloads []*workload.Info
+			for _, target := range targets {
+				targetWorkloads = append(targetWorkloads, target.WorkloadInfo)
+			}
+			revertUsage := cq.SimulateUsageRemoval(targetWorkloads)
+			tasResult = cq.FindTopologyAssignmentsForWorkload(tasRequests, false)
+			revertUsage()
+		} else {
+			// In this scenario we don't have any preemption candidates, yet we need
+			// to reserve the TAS resources to avoid the situation when a lower
+			// priority workload further in the queue gets admitted and preempted
+			// in the next scheduling cycle by the waiting workload. To obtain
+			// a TAS assignment for reserving the resources we run the algorithm
+			// assuming the cluster is empty.
+			tasResult = cq.FindTopologyAssignmentsForWorkload(tasRequests, true)
+		}
+		assignment.UpdateForTASResult(tasResult)
+	}
+}
+
 // admit sets the admitting clusterQueue and flavors into the workload of
 // the entry, and asynchronously updates the object in the apiserver after
 // assuming it in the cache.
@@ -515,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 {
@@ -540,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)
@@ -554,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)
@@ -567,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.
@@ -586,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/pkg/workload/workload.go b/pkg/workload/workload.go
index 966c3c170..8d146a4a6 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -281,17 +281,17 @@ func dropExcludedResources(input corev1.ResourceList, excludedPrefixes []string)
 	return res
 }
 
-// IsUsingTAS returns information if the workload is using TAS
-func (i *Info) IsUsingTAS() bool {
-	return slices.ContainsFunc(i.TotalRequests,
-		func(ps PodSetResources) bool {
+// IsRequestingTAS returns information if the workload is requesting TAS
+func (i *Info) IsRequestingTAS() bool {
+	return slices.ContainsFunc(i.Obj.Spec.PodSets,
+		func(ps kueue.PodSet) bool {
 			return ps.TopologyRequest != nil
 		})
 }
 
 // TASUsage returns topology usage requested by the Workload
 func (i *Info) TASUsage() TASUsage {
-	if !features.Enabled(features.TopologyAwareScheduling) || !i.IsUsingTAS() {
+	if !features.Enabled(features.TopologyAwareScheduling) || !i.IsRequestingTAS() {
 		return nil
 	}
 	result := make(TASUsage, 0)
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/clusterqueue_test.go b/pkg/cache/clusterqueue_test.go
index 87cfdd74a..5f7e1de62 100644
--- a/pkg/cache/clusterqueue_test.go
+++ b/pkg/cache/clusterqueue_test.go
@@ -655,8 +655,8 @@ func TestClusterQueueReadinessWithTAS(t *testing.T) {
 						ResourceQuotaWrapper("example.com/gpu").NominalQuota("5").Append().
 						FlavorQuotas,
 				).Cohort("some-cohort").Obj(),
-			wantReason:  kueue.ClusterQueueActiveReasonNotSupportedWithTopologyAwareScheduling,
-			wantMessage: "Can't admit new workloads: TAS is not supported for cohorts.",
+			wantReason:  kueue.ClusterQueueActiveReasonReady,
+			wantMessage: "Can admit new workloads",
 		},
 		{
 			name: "TAS do not support Preemption",
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/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 bf431d3a2..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().
@@ -5676,3 +6072,1176 @@ func TestScheduleForTASPreemption(t *testing.T) {
 		})
 	}
 }
+
+func TestScheduleForTASCohorts(t *testing.T) {
+	defaultNodeX1 := *testingnode.MakeNode("x1").
+		Label("tas-node", "true").
+		Label(corev1.LabelHostname, "x1").
+		StatusAllocatable(corev1.ResourceList{
+			corev1.ResourceCPU:    resource.MustParse("3"),
+			corev1.ResourceMemory: resource.MustParse("5Gi"),
+			corev1.ResourcePods:   resource.MustParse("10"),
+		}).
+		Ready().
+		Obj()
+	defaultNodeY1 := *testingnode.MakeNode("y1").
+		Label("tas-node", "true").
+		Label(corev1.LabelHostname, "y1").
+		StatusAllocatable(corev1.ResourceList{
+			corev1.ResourceCPU:    resource.MustParse("5"),
+			corev1.ResourceMemory: resource.MustParse("3Gi"),
+			corev1.ResourcePods:   resource.MustParse("10"),
+		}).
+		Ready().
+		Obj()
+	defaultTwoNodes := []corev1.Node{defaultNodeX1, defaultNodeY1}
+	defaultSingleLevelTopology := *utiltesting.MakeDefaultOneLevelTopology("tas-single-level")
+	defaultTASFlavor := *utiltesting.MakeResourceFlavor("tas-default").
+		NodeLabel("tas-node", "true").
+		TopologyName("tas-single-level").
+		Obj()
+	defaultClusterQueueA := *utiltesting.MakeClusterQueue("tas-cq-a").
+		Cohort("tas-cohort-main").
+		Preemption(kueue.ClusterQueuePreemption{
+			WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+			ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+		}).
+		ResourceGroup(*utiltesting.MakeFlavorQuotas("tas-default").
+			Resource(corev1.ResourceCPU, "4").
+			Resource(corev1.ResourceMemory, "4Gi").Obj()).
+		Obj()
+	defaultClusterQueueB := *utiltesting.MakeClusterQueue("tas-cq-b").
+		Cohort("tas-cohort-main").
+		Preemption(kueue.ClusterQueuePreemption{
+			ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+		}).
+		ResourceGroup(*utiltesting.MakeFlavorQuotas("tas-default").
+			Resource(corev1.ResourceCPU, "4").
+			Resource(corev1.ResourceMemory, "4Gi").Obj()).
+		Obj()
+	defaultClusterQueueC := *utiltesting.MakeClusterQueue("tas-cq-c").
+		Cohort("tas-cohort-main").
+		Preemption(kueue.ClusterQueuePreemption{
+			ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+		}).
+		ResourceGroup(*utiltesting.MakeFlavorQuotas("tas-default").
+			Resource(corev1.ResourceCPU, "4").
+			Resource(corev1.ResourceMemory, "4Gi").Obj()).
+		Obj()
+	queues := []kueue.LocalQueue{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "default",
+				Name:      "tas-lq-a",
+			},
+			Spec: kueue.LocalQueueSpec{
+				ClusterQueue: "tas-cq-a",
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "default",
+				Name:      "tas-lq-b",
+			},
+			Spec: kueue.LocalQueueSpec{
+				ClusterQueue: "tas-cq-b",
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "default",
+				Name:      "tas-lq-c",
+			},
+			Spec: kueue.LocalQueueSpec{
+				ClusterQueue: "tas-cq-c",
+			},
+		},
+	}
+	eventIgnoreMessage := cmpopts.IgnoreFields(utiltesting.EventRecord{}, "Message")
+	cases := map[string]struct {
+		nodes           []corev1.Node
+		pods            []corev1.Pod
+		topologies      []kueuealpha.Topology
+		resourceFlavors []kueue.ResourceFlavor
+		clusterQueues   []kueue.ClusterQueue
+		workloads       []kueue.Workload
+
+		// wantNewAssignments is a summary of all new admissions in the cache after this cycle.
+		wantNewAssignments map[string]kueue.Admission
+		// wantLeft is the workload keys that are left in the queues after this cycle.
+		wantLeft map[string][]string
+		// wantInadmissibleLeft is the workload keys that are left in the inadmissible state after this cycle.
+		wantInadmissibleLeft map[string][]string
+		// wantPreempted is the keys of the workloads that get preempted in the scheduling cycle.
+		wantPreempted sets.Set[string]
+		// wantEvents asserts on the events, the comparison options are passed by eventCmpOpts
+		wantEvents []utiltesting.EventRecord
+		// eventCmpOpts are the comparison options for the events
+		eventCmpOpts []cmp.Option
+	}{
+		"workload which requires borrowing gets scheduled": {
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1", "default").
+					Queue("tas-lq-a").
+					PodSets(*utiltesting.MakePodSet("one", 6).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantNewAssignments: map[string]kueue.Admission{
+				"default/a1": *utiltesting.MakeAdmission("tas-cq-a", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "6").
+					AssignmentPodCount(6).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"x1",
+								},
+							},
+							{
+								Count: 5,
+								Values: []string{
+									"y1",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
+		"reclaim within cohort; single borrowing workload gets preempted": {
+			// This is a baseline scenario for reclamation within cohort where
+			// a single borrowing workload gets preempted.
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1-admitted", "default").
+					Queue("tas-lq-a").
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "5").
+							AssignmentPodCount(5).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 5,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 5).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Queue("tas-lq-b").
+					PodSets(*utiltesting.MakePodSet("one", 4).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+			},
+			wantPreempted: sets.New("default/a1-admitted"),
+			eventCmpOpts:  []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1-admitted"},
+					Reason:    "Preempted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
+		"reclaim within cohort; single workload is preempted out three candidates": {
+			// In this scenario the CQA is borrowing and has three candidates.
+			// The test demonstrates the heuristic to select only a subset of
+			// workloads that need to be preempted.
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(1).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "5").
+							AssignmentPodCount(5).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 5,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 5).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("a2-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "2").
+							AssignmentPodCount(2).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 2,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 2).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("a3-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "1").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Queue("tas-lq-b").
+					Priority(3).
+					PodSets(*utiltesting.MakePodSet("one", 3).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+			},
+			wantPreempted: sets.New("default/a1-admitted"),
+			eventCmpOpts:  []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1-admitted"},
+					Reason:    "Preempted",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+			},
+		},
+		"reclaim within cohort; preempting with partial admission": {
+			// The new workload requires 4 units of CPU and memory on a single
+			// node. There is no such node, so it gets schedule after partial
+			// admission to 3 units of CPU and memory. It preempts one workload
+			// to fit.
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(3).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "5").
+							AssignmentPodCount(5).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 5,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 4).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("a2-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "3").
+							AssignmentPodCount(3).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 3,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 3).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Queue("tas-lq-b").
+					Priority(3).
+					PodSets(*utiltesting.MakePodSet("one", 4).
+						SetMinimumCount(3).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Request(corev1.ResourceMemory, "1").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+			},
+			wantPreempted: sets.New("default/a2-admitted"),
+			eventCmpOpts:  []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a2-admitted"},
+					Reason:    "Preempted",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+			},
+		},
+		"reclaim within cohort; capacity reserved by preempting workload does not allow to schedule last workload": {
+			// The c1 workload is initially categorized as Fit, because there
+			// was one CPU unit empty. However, the preempting workload b1 books
+			// this capacity.
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB, defaultClusterQueueC},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "5").
+							AssignmentPodCount(5).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 3,
+										Values: []string{
+											"x1",
+										},
+									},
+									{
+										Count: 2,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 5).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("a2-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(1).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "1").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("a3-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(1).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "1").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Queue("tas-lq-b").
+					Priority(3).
+					PodSets(*utiltesting.MakePodSet("one", 3).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("c1", "default").
+					Queue("tas-lq-c").
+					Priority(1).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+				"tas-cq-c": {"default/c1"},
+			},
+			wantPreempted: sets.New("default/a2-admitted", "default/a3-admitted"),
+			eventCmpOpts:  []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "c1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a2-admitted"},
+					Reason:    "Preempted",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a3-admitted"},
+					Reason:    "Preempted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
+		"two small workloads considered; both get scheduled on different nodes": {
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1", "default").
+					Priority(2).
+					Queue("tas-lq-a").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "5").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Priority(1).
+					Queue("tas-lq-b").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceMemory, "5").
+						Obj()).
+					Obj(),
+			},
+			wantNewAssignments: map[string]kueue.Admission{
+				"default/a1": *utiltesting.MakeAdmission("tas-cq-a", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "5").
+					AssignmentPodCount(1).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"y1",
+								},
+							},
+						},
+					}).Obj(),
+				"default/b1": *utiltesting.MakeAdmission("tas-cq-b", "one").
+					Assignment(corev1.ResourceMemory, "tas-default", "5").
+					AssignmentPodCount(1).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"x1",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
+		"two small workloads considered; both get scheduled on the same node": {
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1", "default").
+					Priority(2).
+					Queue("tas-lq-a").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Priority(1).
+					Queue("tas-lq-b").
+					PodSets(*utiltesting.MakePodSet("one", 2).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantNewAssignments: map[string]kueue.Admission{
+				"default/a1": *utiltesting.MakeAdmission("tas-cq-a", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "1").
+					AssignmentPodCount(1).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"x1",
+								},
+							},
+						},
+					}).Obj(),
+				"default/b1": *utiltesting.MakeAdmission("tas-cq-b", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "2").
+					AssignmentPodCount(2).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 2,
+								Values: []string{
+									"x1",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
+		"two small workloads considered; there is only space for one of them on the initial node": {
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1", "default").
+					Priority(2).
+					Queue("tas-lq-a").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Priority(1).
+					Queue("tas-lq-b").
+					PodSets(*utiltesting.MakePodSet("one", 3).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantNewAssignments: map[string]kueue.Admission{
+				"default/a1": *utiltesting.MakeAdmission("tas-cq-a", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "1").
+					AssignmentPodCount(1).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"x1",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
+		"two workloads considered; there is enough space only for the first": {
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1", "default").
+					Priority(2).
+					Queue("tas-lq-a").
+					PodSets(*utiltesting.MakePodSet("one", 5).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Priority(1).
+					Queue("tas-lq-b").
+					PodSets(*utiltesting.MakePodSet("one", 5).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+			},
+			wantNewAssignments: map[string]kueue.Admission{
+				"default/a1": *utiltesting.MakeAdmission("tas-cq-a", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "5").
+					AssignmentPodCount(5).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 5,
+								Values: []string{
+									"y1",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
+		"two workloads considered; both overlapping in the initial flavor assignment": {
+			// Both of the workloads require 2 CPU units which will make them
+			// both target the x1 node which is smallest in terms of CPU.
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueA, defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1", "default").
+					Priority(2).
+					Queue("tas-lq-a").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "2").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Priority(1).
+					Queue("tas-lq-b").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "2").
+						Obj()).
+					Obj(),
+			},
+			wantNewAssignments: map[string]kueue.Admission{
+				"default/a1": *utiltesting.MakeAdmission("tas-cq-a", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "2").
+					AssignmentPodCount(1).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"x1",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+			},
+		},
+		"preempting workload with targets reserves capacity so that lower priority workload cannot use it": {
+			nodes:           []corev1.Node{defaultNodeY1},
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues: []kueue.ClusterQueue{*utiltesting.MakeClusterQueue("tas-cq-a").
+				Cohort("tas-cohort-main").
+				Preemption(kueue.ClusterQueuePreemption{
+					WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+					ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+				}).
+				ResourceGroup(*utiltesting.MakeFlavorQuotas("tas-default").
+					Resource(corev1.ResourceCPU, "8", "0").
+					Resource(corev1.ResourceMemory, "4Gi", "0").Obj()).
+				Obj(), defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(1).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "2").
+							AssignmentPodCount(2).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 2,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 2).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("a2", "default").
+					Queue("tas-lq-a").
+					Priority(2).
+					PodSets(*utiltesting.MakePodSet("one", 4).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Queue("tas-lq-b").
+					Priority(1).
+					PodSets(*utiltesting.MakePodSet("one", 3).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-a": {"default/a2"},
+				"tas-cq-b": {"default/b1"},
+			},
+			wantPreempted: sets.New("default/a1-admitted"),
+			eventCmpOpts:  []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a2"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a1-admitted"},
+					Reason:    "Preempted",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+			},
+		},
+		"preempting workload without targets reserves capacity so that lower priority workload cannot use it": {
+			nodes:           []corev1.Node{defaultNodeY1},
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues: []kueue.ClusterQueue{*utiltesting.MakeClusterQueue("tas-cq-a").
+				Cohort("tas-cohort-main").
+				Preemption(kueue.ClusterQueuePreemption{
+					WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+					ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+				}).
+				ResourceGroup(*utiltesting.MakeFlavorQuotas("tas-default").
+					Resource(corev1.ResourceCPU, "8", "0").
+					Resource(corev1.ResourceMemory, "4Gi", "0").Obj()).
+				Obj(), defaultClusterQueueB},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("a1-admitted", "default").
+					Queue("tas-lq-a").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-cq-a", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "2").
+							AssignmentPodCount(2).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 2,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 2).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("a2", "default").
+					Queue("tas-lq-a").
+					Priority(2).
+					PodSets(*utiltesting.MakePodSet("one", 4).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("b1", "default").
+					Queue("tas-lq-b").
+					Priority(1).
+					PodSets(*utiltesting.MakePodSet("one", 3).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-cq-b": {"default/b1"},
+			},
+			wantInadmissibleLeft: map[string][]string{
+				"tas-cq-a": {"default/a2"},
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "a2"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "b1"},
+					Reason:    "Pending",
+					EventType: corev1.EventTypeWarning,
+				},
+			},
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			features.SetFeatureGateDuringTest(t, features.TopologyAwareScheduling, true)
+			ctx, _ := utiltesting.ContextWithLog(t)
+
+			clientBuilder := utiltesting.NewClientBuilder().
+				WithLists(
+					&kueue.WorkloadList{Items: tc.workloads},
+					&kueuealpha.TopologyList{Items: tc.topologies},
+					&corev1.PodList{Items: tc.pods},
+					&corev1.NodeList{Items: tc.nodes},
+					&kueue.LocalQueueList{Items: queues}).
+				WithObjects(
+					&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}},
+				)
+			_ = tasindexer.SetupIndexes(ctx, utiltesting.AsIndexer(clientBuilder))
+			cl := clientBuilder.Build()
+			recorder := &utiltesting.EventRecorder{}
+			cqCache := cache.New(cl)
+			qManager := queue.NewManager(cl, cqCache)
+			topologyByName := slices.ToMap(tc.topologies, func(i int) (kueue.TopologyReference, kueuealpha.Topology) {
+				return kueue.TopologyReference(tc.topologies[i].Name), tc.topologies[i]
+			})
+			for _, flavor := range tc.resourceFlavors {
+				cqCache.AddOrUpdateResourceFlavor(&flavor)
+				if flavor.Spec.TopologyName != nil {
+					t := topologyByName[*flavor.Spec.TopologyName]
+					tasCache := cqCache.TASCache()
+					levels := utiltas.Levels(&t)
+					tasFlavorCache := tasCache.NewTASFlavorCache(*flavor.Spec.TopologyName, levels, flavor.Spec.NodeLabels, flavor.Spec.Tolerations)
+					tasCache.Set(kueue.ResourceFlavorReference(flavor.Name), tasFlavorCache)
+				}
+			}
+			for _, cq := range tc.clusterQueues {
+				if err := cqCache.AddClusterQueue(ctx, &cq); err != nil {
+					t.Fatalf("Inserting clusterQueue %s in cache: %v", cq.Name, err)
+				}
+				if err := qManager.AddClusterQueue(ctx, &cq); err != nil {
+					t.Fatalf("Inserting clusterQueue %s in manager: %v", cq.Name, err)
+				}
+				if err := cl.Create(ctx, &cq); err != nil {
+					t.Fatalf("couldn't create the cluster queue: %v", err)
+				}
+			}
+			for _, q := range queues {
+				if err := qManager.AddLocalQueue(ctx, &q); err != nil {
+					t.Fatalf("Inserting queue %s/%s in manager: %v", q.Namespace, q.Name, err)
+				}
+			}
+			initiallyAdmittedWorkloads := sets.New[string]()
+			for _, w := range tc.workloads {
+				if workload.IsAdmitted(&w) {
+					initiallyAdmittedWorkloads.Insert(workload.Key(&w))
+				}
+			}
+			scheduler := New(qManager, cqCache, cl, recorder)
+			gotScheduled := make([]string, 0)
+			var mu sync.Mutex
+			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+				mu.Lock()
+				gotScheduled = append(gotScheduled, workload.Key(w))
+				mu.Unlock()
+				return nil
+			}
+			wg := sync.WaitGroup{}
+			scheduler.setAdmissionRoutineWrapper(routine.NewWrapper(
+				func() { wg.Add(1) },
+				func() { wg.Done() },
+			))
+
+			gotPreempted := sets.New[string]()
+			scheduler.preemptor.OverrideApply(func(_ context.Context, w *kueue.Workload, _, _ string) error {
+				mu.Lock()
+				gotPreempted.Insert(workload.Key(w))
+				mu.Unlock()
+				return nil
+			})
+
+			ctx, cancel := context.WithTimeout(ctx, queueingTimeout)
+			go qManager.CleanUpOnContext(ctx)
+			defer cancel()
+
+			scheduler.schedule(ctx)
+			wg.Wait()
+			snapshot, err := cqCache.Snapshot(ctx)
+			if err != nil {
+				t.Fatalf("unexpected error while building snapshot: %v", err)
+			}
+			if diff := cmp.Diff(tc.wantPreempted, gotPreempted); diff != "" {
+				t.Errorf("Unexpected preemptions (-want,+got):\n%s", diff)
+			}
+			gotAssignments := make(map[string]kueue.Admission)
+			for cqName, c := range snapshot.ClusterQueues() {
+				for name, w := range c.Workloads {
+					if initiallyAdmittedWorkloads.Has(workload.Key(w.Obj)) {
+						continue
+					}
+					switch {
+					case !workload.HasQuotaReservation(w.Obj):
+						t.Fatalf("Workload %s is not admitted by a clusterQueue, but it is found as member of clusterQueue %s in the cache", name, cqName)
+					case string(w.Obj.Status.Admission.ClusterQueue) != cqName:
+						t.Fatalf("Workload %s is admitted by clusterQueue %s, but it is found as member of clusterQueue %s in the cache", name, w.Obj.Status.Admission.ClusterQueue, cqName)
+					default:
+						gotAssignments[name] = *w.Obj.Status.Admission
+					}
+				}
+			}
+			if diff := cmp.Diff(tc.wantNewAssignments, gotAssignments, cmpopts.EquateEmpty()); diff != "" {
+				t.Errorf("Unexpected assigned clusterQueues in cache (-want,+got):\n%s", diff)
+			}
+			qDump := qManager.Dump()
+			if diff := cmp.Diff(tc.wantLeft, qDump, cmpDump...); diff != "" {
+				t.Errorf("Unexpected elements left in the queue (-want,+got):\n%s", diff)
+			}
+			qDumpInadmissible := qManager.DumpInadmissible()
+			if diff := cmp.Diff(tc.wantInadmissibleLeft, qDumpInadmissible, cmpDump...); diff != "" {
+				t.Errorf("Unexpected elements left in inadmissible workloads (-want,+got):\n%s", diff)
+			}
+			if diff := cmp.Diff(tc.wantEvents, recorder.RecordedEvents, append(tc.eventCmpOpts, cmpopts.SortSlices(utiltesting.SortEvents))...); diff != "" {
+				t.Errorf("unexpected events (-want/+got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/test/e2e/singlecluster/pod_test.go b/test/e2e/singlecluster/pod_test.go
index 810263ce0..b8243bc7a 100644
--- a/test/e2e/singlecluster/pod_test.go
+++ b/test/e2e/singlecluster/pod_test.go
@@ -23,6 +23,7 @@ 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"
@@ -518,7 +519,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/tas/appwrapper_test.go b/test/e2e/tas/appwrapper_test.go
index 9c147f944..15fe000d5 100644
--- a/test/e2e/tas/appwrapper_test.go
+++ b/test/e2e/tas/appwrapper_test.go
@@ -97,7 +97,7 @@ var _ = ginkgo.Describe("TopologyAwareScheduling for AppWrapper", func() {
 					Request(extraResource, "1").
 					Limit(extraResource, "1").
 					Suspend(false).
-					Image(util.E2eTestAgnHostImage, []string{"60s"}).
+					Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
 					PodAnnotation(kueuealpha.PodSetPreferredTopologyAnnotation, testing.DefaultRackTopologyLevel).
 					SetTypeMeta().Obj()).
 				Queue(localQueue.Name).
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index 7202bac81..a2e3a6517 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -216,27 +216,6 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 			util.ExpectObjectToBeDeleted(ctx, k8sClient, admissionCheck, true)
 		})
 
-		ginkgo.It("should mark TAS ClusterQueue as inactive if used in cohort", func() {
-			clusterQueue = testing.MakeClusterQueue("cq").
-				ResourceGroup(
-					*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj(),
-				).Cohort("cohort").Obj()
-			gomega.Expect(k8sClient.Create(ctx, clusterQueue)).Should(gomega.Succeed())
-
-			gomega.Eventually(func(g gomega.Gomega) {
-				var updatedCq kueue.ClusterQueue
-				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), &updatedCq)).To(gomega.Succeed())
-				g.Expect(updatedCq.Status.Conditions).Should(gomega.BeComparableTo([]metav1.Condition{
-					{
-						Type:    kueue.ClusterQueueActive,
-						Status:  metav1.ConditionFalse,
-						Reason:  "NotSupportedWithTopologyAwareScheduling",
-						Message: `Can't admit new workloads: TAS is not supported for cohorts.`,
-					},
-				}, util.IgnoreConditionTimestampsAndObservedGeneration))
-			}, util.Timeout, util.Interval).Should(gomega.Succeed())
-		})
-
 		ginkgo.It("should mark TAS ClusterQueue as inactive if used with MultiKueue", func() {
 			admissionCheck = testing.MakeAdmissionCheck("multikueue").ControllerName(kueue.MultiKueueControllerName).Obj()
 			gomega.Expect(k8sClient.Create(ctx, admissionCheck)).To(gomega.Succeed())
@@ -956,6 +935,137 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 			})
 		})
 
+		ginkgo.When("Preemption is enabled within Cohort", func() {
+			var (
+				nodes         []corev1.Node
+				localQueueB   *kueue.LocalQueue
+				clusterQueueB *kueue.ClusterQueue
+			)
+			ginkgo.BeforeEach(func() {
+				nodes = []corev1.Node{
+					*testingnode.MakeNode("x1").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r1").
+						Label(corev1.LabelHostname, "x1").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("5"),
+							corev1.ResourceMemory: resource.MustParse("5Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+					*testingnode.MakeNode("x2").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r2").
+						Label(corev1.LabelHostname, "x2").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("5"),
+							corev1.ResourceMemory: resource.MustParse("5Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+				}
+				util.CreateNodesWithStatus(ctx, k8sClient, nodes)
+
+				topology = testing.MakeDefaultThreeLevelTopology("default")
+				gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
+
+				tasFlavor = testing.MakeResourceFlavor("tas-flavor").
+					NodeLabel("node-group", "tas").
+					TopologyName("default").Obj()
+				gomega.Expect(k8sClient.Create(ctx, tasFlavor)).Should(gomega.Succeed())
+
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
+					Cohort("cohort").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).
+						Resource(corev1.ResourceCPU, "5").
+						Resource(corev1.ResourceMemory, "5Gi").Obj()).
+					Obj()
+				gomega.Expect(k8sClient.Create(ctx, clusterQueue)).Should(gomega.Succeed())
+				util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQueue)
+
+				localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+				gomega.Expect(k8sClient.Create(ctx, localQueue)).Should(gomega.Succeed())
+
+				clusterQueueB = testing.MakeClusterQueue("cluster-queue-b").
+					Cohort("cohort").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).
+						Resource(corev1.ResourceCPU, "5").
+						Resource(corev1.ResourceMemory, "5Gi").Obj()).
+					Obj()
+				gomega.Expect(k8sClient.Create(ctx, clusterQueueB)).Should(gomega.Succeed())
+				util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQueueB)
+
+				localQueueB = testing.MakeLocalQueue("local-queue-b", ns.Name).ClusterQueue(clusterQueueB.Name).Obj()
+				gomega.Expect(k8sClient.Create(ctx, localQueueB)).Should(gomega.Succeed())
+			})
+
+			ginkgo.AfterEach(func() {
+				gomega.Expect(util.DeleteAllJobsInNamespace(ctx, k8sClient, ns)).Should(gomega.Succeed())
+				gomega.Expect(util.DeleteWorkloadsInNamespace(ctx, k8sClient, ns)).Should(gomega.Succeed())
+				gomega.Expect(util.DeleteObject(ctx, k8sClient, localQueue)).Should(gomega.Succeed())
+				gomega.Expect(util.DeleteObject(ctx, k8sClient, localQueueB)).Should(gomega.Succeed())
+
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueueB, true)
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
+				for _, node := range nodes {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, &node, true)
+				}
+			})
+
+			ginkgo.It("should allow to borrow within cohort and reclaim the capacity", func() {
+				var wl1, wl2 *kueue.Workload
+				ginkgo.By("creating a workload which can fit only when borrowing quota", func() {
+					wl1 = testing.MakeWorkload("wl1", ns.Name).
+						PodSets(*testing.MakePodSet("worker", 2).
+							PreferredTopologyRequest(testing.DefaultBlockTopologyLevel).
+							Obj()).
+						Queue(localQueue.Name).Request(corev1.ResourceCPU, "4").Obj()
+					gomega.Expect(k8sClient.Create(ctx, wl1)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the workload is admitted", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 1)
+				})
+
+				ginkgo.By("creating a workload in CQB which reclaims its quota", func() {
+					// Note that workload wl2 would still fit within regular quota
+					// without preempting wl1. However, there is only 1CPU left
+					// on both nodes, and so wl1 needs to be preempted  to allow
+					// scheduling of wl2.
+					wl2 = testing.MakeWorkload("wl2", ns.Name).
+						Priority(2).
+						PodSets(*testing.MakePodSet("worker", 1).
+							PreferredTopologyRequest(testing.DefaultBlockTopologyLevel).
+							Obj()).
+						Queue(localQueueB.Name).Request(corev1.ResourceCPU, "2").Obj()
+					gomega.Expect(k8sClient.Create(ctx, wl2)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the wl1 gets preempted and wl2 gets admitted", func() {
+					util.ExpectWorkloadsToBePreempted(ctx, k8sClient, wl1)
+					util.FinishEvictionForWorkloads(ctx, k8sClient, wl1)
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl2)
+					util.ExpectReservingActiveWorkloadsMetric(clusterQueueB, 1)
+					util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 0)
+				})
+			})
+		})
+
 		ginkgo.When("Node structure is mutated during test cases", func() {
 			var (
 				nodes []corev1.Node
diff --git a/test/util/constants.go b/test/util/constants.go
index 216abfe73..61c6934d9 100644
--- a/test/util/constants.go
+++ b/test/util/constants.go
@@ -34,7 +34,7 @@ const (
 	LongTimeout = 45 * time.Second
 	// VeryLongTimeout is meant for E2E tests involving Ray which starts ray-project images (over 2GB)
 	// and also synchronizes the cluster before it can be used
-	VeryLongTimeout = 3 * time.Minute
+	VeryLongTimeout = 5 * time.Minute
 	// StartUpTimeout is meant to be used for waiting for Kueue to startup, given
 	// that cert updates can take up to 3 minutes to propagate to the filesystem.
 	// Taken into account that after the certificates are ready, all Kueue's components
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