Repository: kubernetes-sigs/kueue

Issue:
single inadmissible workload from quota never gets pending condition with best effort fifo
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

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


**What happened**:

While working on the integration test for https://github.com/kubernetes-sigs/kueue/issues/4934 // https://github.com/kubernetes-sigs/kueue/pull/4935 I noticed a curious issue

Workloads never get any conditions set, but are actively being hit by both the scheduler and the reconciler. The culprit seems to be a resource version conflict between scheduler and reconciler such that the scheduler never applies this status update during failed admission of head entries: https://github.com/kubernetes-sigs/kueue/blob/3279d9c05817e465229fac6bdc64250c890ea7dd/pkg/scheduler/scheduler.go#L659

resulting in e.g.
```
  2025-04-21T17:49:24.03698-04:00	ERROR	scheduler	scheduler/scheduler.go:685	Could not update Workload status	{"schedulingCycle": 5, "error": "Operation cannot be fulfilled on workloads.kueue.x-k8s.io \"admission-check-wl2\": the object has been modified; please apply your changes to the latest version and try again"}
  sigs.k8s.io/kueue/pkg/scheduler.(*Scheduler).requeueAndUpdate
  	/Users/alexeldeib/code/kueue/pkg/scheduler/scheduler.go:685
  sigs.k8s.io/kueue/pkg/scheduler.(*Scheduler).schedule
  	/Users/alexeldeib/code/kueue/pkg/scheduler/scheduler.go:302
  sigs.k8s.io/kueue/pkg/util/wait.untilWithBackoff.func1
  	/Users/alexeldeib/code/kueue/pkg/util/wait/backoff.go:43
  k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1
  	/Users/alexeldeib/code/kueue/vendor/k8s.io/apimachinery/pkg/util/wait/backoff.go:226
  k8s.io/apimachinery/pkg/util/wait.BackoffUntil
  	/Users/alexeldeib/code/kueue/vendor/k8s.io/apimachinery/pkg/util/wait/backoff.go:227
  sigs.k8s.io/kueue/pkg/util/wait.untilWithBackoff
  	/Users/alexeldeib/code/kueue/pkg/util/wait/backoff.go:42
  sigs.k8s.io/kueue/pkg/util/wait.UntilWithBackoff
  	/Users/alexeldeib/code/kueue/pkg/util/wait/backoff.go:34
```

in this case it seems there is no workload reconcile, but the workload is requeued as inadmissible and never re-scheduled/nominated when it hits https://github.com/kubernetes-sigs/kueue/blob/3279d9c05817e465229fac6bdc64250c890ea7dd/pkg/queue/cluster_queue.go#L406

OR the update passes, but the workload reconciler triggers a no-op update from pending to pending. see this code path https://github.com/kubernetes-sigs/kueue/blob/3279d9c05817e465229fac6bdc64250c890ea7dd/pkg/controller/core/workload_controller.go#L704-L705

either case ends up with the workload requeued as inadmissible, and then it may never get requeued. there is no reason a single workload would retrigger scheduling once it is inadmissible, unless other workloads are deleted, or the CQs are updated, etc.

there are two potential fixes which seem to both be required:
- trigger requeue of inadmissible workload immediately on resource version conflict (e.g. `apierrors.IsConflict`) during requeue status update
  - this solves the first case, since without immediate requeue and no additional update from workload controller, it's kaput
- trigger requeue of inadmissible workloads during the pending -> pending reconciler in workload controller as well as the default path (for spurious/uncached events).
  - this handles the case where the status update succeeds, triggers a workload reconcile, but that does not currently retrigger a scheduling loop

**What you expected to happen**:

condition status updates should occur on pending workloads

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

see https://github.com/kubernetes-sigs/kueue/pull/4935 -- remove the changes mentioned above and run the test added in that PR a few times, it will reproduce both variations.

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

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

Discussion:
@alexeldeib: /assign@alexeldeib: tracking this as a separate issue for viz in case anyone else hits it since it's ultimately different and #4935 has drifted quite a bit from the original 1-line fix for #4934, but it seems like a fix for #4934 may require fixing this issue as well to get the correct status update behavior@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle rotten

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@alexeldeib: I don’t think we’ve managed to reproduce this since upgrading to 0.12

/close

On Wed, Aug 20, 2025 at 1:10 PM Kubernetes Triage Robot <
***@***.***> wrote:

> *k8s-triage-robot* left a comment (kubernetes-sigs/kueue#5061)
> <https://github.com/kubernetes-sigs/kueue/issues/5061#issuecomment-3207324056>
>
> The Kubernetes project currently lacks enough active contributors to
> adequately respond to all issues.
>
> This bot triages un-triaged issues according to the following rules:
>
>    - After 90d of inactivity, lifecycle/stale is applied
>    - After 30d of inactivity since lifecycle/stale was applied,
>    lifecycle/rotten is applied
>    - After 30d of inactivity since lifecycle/rotten was applied, the
>    issue is closed
>
> You can:
>
>    - Mark this issue as fresh with /remove-lifecycle rotten
>    - Close this issue with /close
>    - Offer to help out with Issue Triage
>    <https://www.kubernetes.dev/docs/guide/issue-triage/>
>
> Please send feedback to sig-contributor-experience at kubernetes/community
> <https://github.com/kubernetes/community>.
>
> /lifecycle rotten
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/kubernetes-sigs/kueue/issues/5061#issuecomment-3207324056>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/ABT4LWKUCIJMOVCUO6UIHBD3OSTXVAVCNFSM6AAAAAB3SL43D2VHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZTEMBXGMZDIMBVGY>
> .
> You are receiving this because you were assigned.Message ID:
> ***@***.***>
>
@k8s-ci-robot: @alexeldeib: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/5061#issuecomment-3207720985):

>I don’t think we’ve managed to reproduce this since upgrading to 0.12
>
>/close
>
>On Wed, Aug 20, 2025 at 1:10 PM Kubernetes Triage Robot <
>***@***.***> wrote:
>
>> *k8s-triage-robot* left a comment (kubernetes-sigs/kueue#5061)
>> <https://github.com/kubernetes-sigs/kueue/issues/5061#issuecomment-3207324056>
>>
>> The Kubernetes project currently lacks enough active contributors to
>> adequately respond to all issues.
>>
>> This bot triages un-triaged issues according to the following rules:
>>
>>    - After 90d of inactivity, lifecycle/stale is applied
>>    - After 30d of inactivity since lifecycle/stale was applied,
>>    lifecycle/rotten is applied
>>    - After 30d of inactivity since lifecycle/rotten was applied, the
>>    issue is closed
>>
>> You can:
>>
>>    - Mark this issue as fresh with /remove-lifecycle rotten
>>    - Close this issue with /close
>>    - Offer to help out with Issue Triage
>>    <https://www.kubernetes.dev/docs/guide/issue-triage/>
>>
>> Please send feedback to sig-contributor-experience at kubernetes/community
>> <https://github.com/kubernetes/community>.
>>
>> /lifecycle rotten
>>
>> —
>> Reply to this email directly, view it on GitHub
>> <https://github.com/kubernetes-sigs/kueue/issues/5061#issuecomment-3207324056>,
>> or unsubscribe
>> <https://github.com/notifications/unsubscribe-auth/ABT4LWKUCIJMOVCUO6UIHBD3OSTXVAVCNFSM6AAAAAB3SL43D2VHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZTEMBXGMZDIMBVGY>
>> .
>> You are receiving this because you were assigned.Message ID:
>> ***@***.***>
>>
>


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

Solution patch:
diff --git a/pkg/cache/cohort_snapshot.go b/pkg/cache/cohort_snapshot.go
index 5dd7c8bf1..bf74056e6 100644
--- a/pkg/cache/cohort_snapshot.go
+++ b/pkg/cache/cohort_snapshot.go
@@ -21,6 +21,7 @@ import (
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/hierarchy"
+	"sigs.k8s.io/kueue/pkg/resources"
 )
 
 type CohortSnapshot struct {
@@ -88,3 +89,7 @@ func (c *CohortSnapshot) parentHRN() hierarchicalResourceNode {
 func (c *CohortSnapshot) fairWeight() *resource.Quantity {
 	return &c.FairWeight
 }
+
+func (c *CohortSnapshot) BorrowingWith(fr resources.FlavorResource, val int64) bool {
+	return c.ResourceNode.Usage[fr]+val > c.ResourceNode.SubtreeQuota[fr]
+}
diff --git a/pkg/cache/resource_node.go b/pkg/cache/resource_node.go
index 650205116..9d4062798 100644
--- a/pkg/cache/resource_node.go
+++ b/pkg/cache/resource_node.go
@@ -19,6 +19,8 @@ package cache
 import (
 	"maps"
 
+	"k8s.io/apimachinery/pkg/util/sets"
+
 	"sigs.k8s.io/kueue/pkg/hierarchy"
 	"sigs.k8s.io/kueue/pkg/resources"
 )
@@ -77,6 +79,13 @@ type hierarchicalResourceNode interface {
 	parentHRN() hierarchicalResourceNode
 }
 
+// LocalAvailable returns, for a given node and resource flavor,
+// how much guaranteed quota in this flavor exceeds usage.
+// This quota is available at this node but is not visible at its parent.
+func LocalAvailable(node hierarchicalResourceNode, fr resources.FlavorResource) int64 {
+	return max(0, node.getResourceNode().guaranteedQuota(fr)-node.getResourceNode().Usage[fr])
+}
+
 // available determines how much capacity remains for the current
 // node, taking into account usage and BorrowingLimits. It finds remaining
 // capacity which is stored locally. If the node has a parent, it
@@ -91,7 +100,6 @@ func available(node hierarchicalResourceNode, fr resources.FlavorResource) int64
 	if !node.HasParent() {
 		return r.SubtreeQuota[fr] - r.Usage[fr]
 	}
-	localAvailable := max(0, r.guaranteedQuota(fr)-r.Usage[fr])
 	parentAvailable := available(node.parentHRN(), fr)
 
 	if borrowingLimit := r.Quotas[fr].BorrowingLimit; borrowingLimit != nil {
@@ -100,7 +108,7 @@ func available(node hierarchicalResourceNode, fr resources.FlavorResource) int64
 		withMaxFromParent := storedInParent - usedInParent + *borrowingLimit
 		parentAvailable = min(withMaxFromParent, parentAvailable)
 	}
-	return localAvailable + parentAvailable
+	return LocalAvailable(node, fr) + parentAvailable
 }
 
 // potentialAvailable returns the maximum capacity available to this node,
@@ -122,7 +130,7 @@ func potentialAvailable(node hierarchicalResourceNode, fr resources.FlavorResour
 // its Cohort when usage exceeds guaranteedQuota.
 func addUsage(node hierarchicalResourceNode, fr resources.FlavorResource, val int64) {
 	r := node.getResourceNode()
-	localAvailable := max(0, r.guaranteedQuota(fr)-r.Usage[fr])
+	localAvailable := LocalAvailable(node, fr)
 	r.Usage[fr] += val
 	if node.HasParent() && val > localAvailable {
 		deltaParentUsage := val - localAvailable
@@ -191,3 +199,30 @@ func accumulateFromChild(parent *cohort, child hierarchicalResourceNode) {
 		parent.resourceNode.Usage[fr] += max(0, childUsage-child.getResourceNode().guaranteedQuota(fr))
 	}
 }
+
+// QuantitiesFitInQuota returns if resource requests fit in quota on this node
+// and the amount of resources that exceed the guaranteed
+// quota on this node. It is assumed that subsequent call
+// to this function will be on nodes parent with remainingRequests
+func QuantitiesFitInQuota(node hierarchicalResourceNode, requests resources.FlavorResourceQuantities) (bool, resources.FlavorResourceQuantities) {
+	fits := true
+	remainingRequests := make(resources.FlavorResourceQuantities, len(requests))
+	for fr, v := range requests {
+		if node.getResourceNode().Usage[fr]+v > node.getResourceNode().SubtreeQuota[fr] {
+			fits = false
+		}
+		remainingRequests[fr] = max(0, v-LocalAvailable(node, fr))
+	}
+	return fits, remainingRequests
+}
+
+// IsWithinNominalInResources returns whether or not, the node quota usage exceeds its
+// nominal quota in any resource flavor out of a set of resource flavours.
+func IsWithinNominalInResources(node hierarchicalResourceNode, frs sets.Set[resources.FlavorResource]) bool {
+	for fr := range frs {
+		if node.getResourceNode().Usage[fr] > node.getResourceNode().SubtreeQuota[fr] {
+			return false
+		}
+	}
+	return true
+}
diff --git a/pkg/scheduler/flavorassigner/flavorassigner.go b/pkg/scheduler/flavorassigner/flavorassigner.go
index 5fd15714e..0000fd4ce 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner.go
@@ -37,12 +37,15 @@ import (
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/resources"
+	"sigs.k8s.io/kueue/pkg/scheduler/preemption/classical"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
 type Assignment struct {
-	PodSets   []PodSetAssignment
-	Borrowing bool
+	PodSets []PodSetAssignment
+	// Borrowing is the height of the smallest cohort tree that fits
+	// the additional Usage. It equals to 0 if no borrowing is required.
+	Borrowing int
 	LastState workload.AssignmentClusterQueueState
 
 	// Usage is the accumulated Usage of resources as pod sets get
@@ -86,7 +89,7 @@ func (a *Assignment) computeTASUsage() workload.TASUsage {
 }
 
 // Borrows return whether assignment requires borrowing.
-func (a *Assignment) Borrows() bool {
+func (a *Assignment) Borrows() int {
 	return a.Borrowing
 }
 
@@ -331,7 +334,7 @@ type FlavorAssignment struct {
 	Name           kueue.ResourceFlavorReference
 	Mode           FlavorAssignmentMode
 	TriedFlavorIdx int
-	borrow         bool
+	borrow         int
 }
 
 type preemptionOracle interface {
@@ -481,8 +484,8 @@ func (a *Assignment) append(requests resources.Requests, psAssignment *PodSetAss
 	flavorIdx := make(map[corev1.ResourceName]int, len(psAssignment.Flavors))
 	a.PodSets = append(a.PodSets, *psAssignment)
 	for resource, flvAssignment := range psAssignment.Flavors {
-		if flvAssignment.borrow {
-			a.Borrowing = true
+		if flvAssignment.borrow > a.Borrowing {
+			a.Borrowing = flvAssignment.borrow
 		}
 		fr := resources.FlavorResource{Flavor: flvAssignment.Name, Resource: resource}
 		a.Usage.Quota[fr] += requests[resource]
@@ -568,7 +571,7 @@ func (a *FlavorAssigner) findFlavorForPodSetResource(
 			if mode < representativeMode {
 				representativeMode = mode
 			}
-			needsBorrowing = needsBorrowing || borrow
+			needsBorrowing = needsBorrowing || (borrow > 0)
 			if representativeMode == noFit {
 				// The flavor doesn't fit, no need to check other resources.
 				break
@@ -689,10 +692,9 @@ func flavorSelector(spec *corev1.PodSpec, allowedKeys sets.Set[string]) nodeaffi
 // if borrowing is required when preempting.
 // If the flavor doesn't satisfy limits immediately (when waiting or preemption
 // could help), it returns a Status with reasons.
-func (a *FlavorAssigner) fitsResourceQuota(log logr.Logger, fr resources.FlavorResource, val int64, rQuota cache.ResourceQuota) (granularMode, bool, *Status) {
+func (a *FlavorAssigner) fitsResourceQuota(log logr.Logger, fr resources.FlavorResource, val int64, rQuota cache.ResourceQuota) (granularMode, int, *Status) {
 	var status Status
 
-	borrow := a.cq.BorrowingWith(fr, val) && a.cq.HasParent()
 	available := a.cq.Available(fr)
 	maxCapacity := a.cq.PotentialAvailable(fr)
 
@@ -700,9 +702,10 @@ func (a *FlavorAssigner) fitsResourceQuota(log logr.Logger, fr resources.FlavorR
 	if val > maxCapacity {
 		status.appendf("insufficient quota for %s in flavor %s, request > maximum capacity (%s > %s)",
 			fr.Resource, fr.Flavor, resources.ResourceQuantityString(fr.Resource, val), resources.ResourceQuantityString(fr.Resource, maxCapacity))
-		return noFit, false, &status
+		return noFit, 0, &status
 	}
 
+	borrow, mayReclaimInHierarchy := classical.FindHeightOfLowestSubtreeThatFits(a.cq, fr, val)
 	// Fit
 	if val <= available {
 		return fit, borrow, nil
@@ -710,7 +713,8 @@ func (a *FlavorAssigner) fitsResourceQuota(log logr.Logger, fr resources.FlavorR
 
 	// Check if preemption is possible
 	mode := noFit
-	if val <= rQuota.Nominal {
+	// For single-level hierarchies, mayReclaimInHierarchy = true iff val <= rQuota.Nominal
+	if val <= rQuota.Nominal || mayReclaimInHierarchy {
 		mode = preempt
 		if a.oracle.IsReclaimPossible(log, a.cq, *a.wl, fr, val) {
 			mode = reclaim
diff --git a/pkg/scheduler/preemption/classical/candidate_generator.go b/pkg/scheduler/preemption/classical/candidate_generator.go
new file mode 100644
index 000000000..c782a5520
--- /dev/null
+++ b/pkg/scheduler/preemption/classical/candidate_generator.go
@@ -0,0 +1,160 @@
+/*
+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 classical
+
+import (
+	"sort"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/utils/clock"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/resources"
+	"sigs.k8s.io/kueue/pkg/workload"
+)
+
+type candidateIterator struct {
+	candidates                        []*candidateElem
+	runIndex                          int
+	frsNeedPreemption                 sets.Set[resources.FlavorResource]
+	snapshot                          *cache.Snapshot
+	NoCandidateFromOtherQueues        bool
+	NoCandidateForHierarchicalReclaim bool
+	hierarchicalReclaimCtx            *HierarchicalPreemptionCtx
+}
+
+type candidateElem struct {
+	wl *workload.Info
+	// lca of this queue and cq (queue to which the new workload is submitted)
+	lca *cache.CohortSnapshot
+	// candidates above priority threshold cannot be preempted if at the same time
+	// cq would borrow from other queues/cohorts
+	preemptionVariant preemptionVariant
+}
+
+// Need a separate function for candidateElem data type
+// in the future we will modify the ordering to take into
+// account the distance in the cohort hierarchy tree
+func candidateElemsOrdering(candidates []*candidateElem, cq kueue.ClusterQueueReference, now time.Time, ordering func([]*workload.Info, kueue.ClusterQueueReference, time.Time) func(int, int) bool) func(int, int) bool {
+	// Adapt the candidatesOrdering function to work with candidateElem
+	adaptedOrdering := func(i, j int) bool {
+		a := candidates[i].wl
+		b := candidates[j].wl
+		return ordering([]*workload.Info{a, b}, cq, now)(0, 1)
+	}
+	return adaptedOrdering
+}
+
+func WorkloadUsesResources(wl *workload.Info, frsNeedPreemption sets.Set[resources.FlavorResource]) bool {
+	for _, ps := range wl.TotalRequests {
+		for res, flv := range ps.Flavors {
+			if frsNeedPreemption.Has(resources.FlavorResource{Flavor: flv, Resource: res}) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// assume a prefix of the elements has condition WorkloadEvicted = true
+func splitEvicted(workloads []*candidateElem) ([]*candidateElem, []*candidateElem) {
+	firstFalse := sort.Search(len(workloads), func(i int) bool {
+		return !meta.IsStatusConditionTrue(workloads[i].wl.Obj.Status.Conditions, kueue.WorkloadEvicted)
+	})
+	return workloads[:firstFalse], workloads[firstFalse:]
+}
+
+// NewCandidateIterator creates a new iterator that yields candidate workloads for preemption
+// The iterator can be used to perform two independent runs over the list of candidates:
+// with and without borrowing. The runs are independent which means that the same candidates
+// might be returned for both, but note that the candidates with borrrowing are a subset of
+// candidates without borrowing.
+func NewCandidateIterator(hierarchicalReclaimCtx *HierarchicalPreemptionCtx, frsNeedPreemption sets.Set[resources.FlavorResource], snapshot *cache.Snapshot, clock clock.Clock, ordering func([]*workload.Info, kueue.ClusterQueueReference, time.Time) func(int, int) bool) *candidateIterator {
+	sameQueueCandidates := collectSameQueueCandidates(hierarchicalReclaimCtx)
+	hierarchyCandidates, priorityCandidates := collectCandidatesForHierarchicalReclaim(hierarchicalReclaimCtx)
+	sort.Slice(sameQueueCandidates, candidateElemsOrdering(sameQueueCandidates, hierarchicalReclaimCtx.Cq.Name, clock.Now(), ordering))
+	sort.Slice(priorityCandidates, candidateElemsOrdering(priorityCandidates, hierarchicalReclaimCtx.Cq.Name, clock.Now(), ordering))
+	sort.Slice(hierarchyCandidates, candidateElemsOrdering(hierarchyCandidates, hierarchicalReclaimCtx.Cq.Name, clock.Now(), ordering))
+	evictedHierarchicalReclaimCandidates, nonEvictedHierarchicalReclaimCandidates := splitEvicted(hierarchyCandidates)
+	evictedSTCandidates, nonEvictedSTCandidates := splitEvicted(priorityCandidates)
+	evictedSameQueueCandidates, nonEvictedSameQueueCandidates := splitEvicted(sameQueueCandidates)
+	allCandidates := make([]*candidateElem, 0, len(hierarchyCandidates)+len(priorityCandidates)+len(sameQueueCandidates))
+	allCandidates = append(allCandidates, evictedHierarchicalReclaimCandidates...)
+	allCandidates = append(allCandidates, evictedSTCandidates...)
+	allCandidates = append(allCandidates, evictedSameQueueCandidates...)
+	allCandidates = append(allCandidates, nonEvictedHierarchicalReclaimCandidates...)
+	allCandidates = append(allCandidates, nonEvictedSTCandidates...)
+	allCandidates = append(allCandidates, nonEvictedSameQueueCandidates...)
+	return &candidateIterator{
+		runIndex:                          0,
+		frsNeedPreemption:                 frsNeedPreemption,
+		snapshot:                          snapshot,
+		candidates:                        allCandidates,
+		NoCandidateFromOtherQueues:        len(hierarchyCandidates) == 0 && len(priorityCandidates) == 0,
+		NoCandidateForHierarchicalReclaim: len(hierarchyCandidates) == 0,
+		hierarchicalReclaimCtx:            hierarchicalReclaimCtx,
+	}
+}
+
+// Next allows to iterate over the ordered sequence of candidates, with the reason
+// for eviction returned together with a candidate.
+func (c *candidateIterator) Next(borrow bool) (*workload.Info, string) {
+	if c.runIndex >= len(c.candidates) {
+		return nil, ""
+	}
+	candidate := c.candidates[c.runIndex]
+	c.runIndex++
+	if !c.candidateIsValid(candidate, borrow) {
+		return c.Next(borrow)
+	}
+	return candidate.wl, candidate.preemptionVariant.PreemptionReason()
+}
+
+// candidateIsValid checks if candidate is valid,
+// as eg. some candidates can only be considered without borrowing
+// Also, preemption of candidates might invalidate other candidates
+func (c *candidateIterator) candidateIsValid(candidate *candidateElem, borrow bool) bool {
+	if c.hierarchicalReclaimCtx.Cq.Name == candidate.wl.ClusterQueue {
+		return true
+	}
+	if borrow && candidate.preemptionVariant == ReclaimWithoutBorrowing {
+		return false
+	}
+	cq := c.snapshot.ClusterQueue(candidate.wl.ClusterQueue)
+	if cache.IsWithinNominalInResources(cq, c.frsNeedPreemption) {
+		return false
+	}
+	// we don't go all the way to the root but only to the lca node
+	for node := range cq.PathParentToRoot() {
+		if node == candidate.lca {
+			break
+		}
+		if cache.IsWithinNominalInResources(node, c.frsNeedPreemption) {
+			return false
+		}
+	}
+	return true
+}
+
+// Reset moves the candidate iterator back to the starting position.
+// It is required to reset the iterator before each run.
+func (c *candidateIterator) Reset() {
+	c.runIndex = 0
+}
diff --git a/pkg/scheduler/preemption/classical/hierarchical_preemption.go b/pkg/scheduler/preemption/classical/hierarchical_preemption.go
new file mode 100644
index 000000000..80d0aa17f
--- /dev/null
+++ b/pkg/scheduler/preemption/classical/hierarchical_preemption.go
@@ -0,0 +1,234 @@
+/*
+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 classical
+
+import (
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/resources"
+	"sigs.k8s.io/kueue/pkg/util/priority"
+	"sigs.k8s.io/kueue/pkg/workload"
+)
+
+type preemptionVariant int
+
+const (
+	// Cannot be preempted
+	Never preemptionVariant = iota
+	// Candidate within the same CQ as the preemptor
+	WithinCQ
+	// Preemptor has preferential access to the resources needing preemption
+	// over the candidate, because of its CQ position in the cohort topology.
+	HiearchicalReclaim
+	// Can only be preempted if preemptor CQ (after all preemptions and the
+	// admission of the incoming workload) would not be borrowing any quota
+	ReclaimWithoutBorrowing
+	// Can be preemped even if preemptor CQ would be borrowing
+	ReclaimWhileBorrowing
+)
+
+func (m preemptionVariant) PreemptionReason() string {
+	switch m {
+	case WithinCQ:
+		return kueue.InClusterQueueReason
+	case HiearchicalReclaim:
+		return kueue.InCohortReclamationReason
+	case ReclaimWhileBorrowing:
+		return kueue.InCohortReclaimWhileBorrowingReason
+	case ReclaimWithoutBorrowing:
+		return kueue.InCohortReclamationReason
+	}
+	return "Unknown"
+}
+
+type HierarchicalPreemptionCtx struct {
+	Wl                *kueue.Workload
+	Cq                *cache.ClusterQueueSnapshot
+	FrsNeedPreemption sets.Set[resources.FlavorResource]
+	Requests          resources.FlavorResourceQuantities
+	WorkloadOrdering  workload.Ordering
+}
+
+func IsBorrowingWithinCohortForbidden(cq *cache.ClusterQueueSnapshot) (bool, *int32) {
+	borrowWithinCohort := cq.Preemption.BorrowWithinCohort
+	if borrowWithinCohort == nil || borrowWithinCohort.Policy == kueue.BorrowWithinCohortPolicyNever {
+		return true, nil
+	}
+	return false, borrowWithinCohort.MaxPriorityThreshold
+}
+
+// classifyPreemptionVariant evaluates, based on config and priorities, the
+// preemption type for a given candidate
+func classifyPreemptionVariant(ctx *HierarchicalPreemptionCtx, wl *workload.Info, haveHierarchicalAdvantage bool) preemptionVariant {
+	if !WorkloadUsesResources(wl, ctx.FrsNeedPreemption) {
+		return Never
+	}
+	incomingPriority := priority.Priority(ctx.Wl)
+	candidatePriority := priority.Priority(wl.Obj)
+	if !satisfiesPreemptionPolicy(ctx, wl, incomingPriority, candidatePriority) {
+		return Never
+	}
+	if wl.ClusterQueue == ctx.Cq.Name {
+		return WithinCQ
+	}
+	if haveHierarchicalAdvantage {
+		return HiearchicalReclaim
+	}
+	borrowWithinCohortForbidden, borrowWithinCohortThreshold := IsBorrowingWithinCohortForbidden(ctx.Cq)
+	if borrowWithinCohortForbidden {
+		return ReclaimWithoutBorrowing
+	}
+	if isAboveBorrowingThreshold(candidatePriority, incomingPriority, borrowWithinCohortThreshold) {
+		return ReclaimWithoutBorrowing
+	}
+	return ReclaimWhileBorrowing
+}
+
+func satisfiesPreemptionPolicy(ctx *HierarchicalPreemptionCtx, wl *workload.Info, incomingPriority, candidatePriority int32) bool {
+	var preemptionPolicy kueue.PreemptionPolicy
+	if wl.ClusterQueue == ctx.Cq.Name {
+		preemptionPolicy = ctx.Cq.Preemption.WithinClusterQueue
+	} else {
+		preemptionPolicy = ctx.Cq.Preemption.ReclaimWithinCohort
+	}
+	lowerPriority := incomingPriority > candidatePriority
+	if preemptionPolicy == kueue.PreemptionPolicyLowerPriority {
+		return lowerPriority
+	}
+	if preemptionPolicy == kueue.PreemptionPolicyLowerOrNewerEqualPriority {
+		preemptorTS := ctx.WorkloadOrdering.GetQueueOrderTimestamp(ctx.Wl)
+		newerEqualPriority := (incomingPriority == candidatePriority) && preemptorTS.Before(ctx.WorkloadOrdering.GetQueueOrderTimestamp(wl.Obj))
+		return (lowerPriority || newerEqualPriority)
+	}
+	return preemptionPolicy == kueue.PreemptionPolicyAny
+}
+
+func isAboveBorrowingThreshold(candidatePriority, incomingPriority int32, borrowWithinCohortThreshold *int32) bool {
+	if candidatePriority >= incomingPriority {
+		return true
+	}
+	if borrowWithinCohortThreshold == nil {
+		return false
+	}
+	return candidatePriority > *borrowWithinCohortThreshold
+}
+
+func collectSameQueueCandidates(ctx *HierarchicalPreemptionCtx) []*candidateElem {
+	if ctx.Cq.Preemption.WithinClusterQueue == kueue.PreemptionPolicyNever {
+		return []*candidateElem{}
+	}
+	return getCandidatesFromCQ(ctx.Cq, nil, ctx, false)
+}
+
+func getCandidatesFromCQ(cq *cache.ClusterQueueSnapshot, lca *cache.CohortSnapshot, ctx *HierarchicalPreemptionCtx, hasHiearchicalAdvantage bool) []*candidateElem {
+	candidates := []*candidateElem{}
+	for _, candidateWl := range cq.Workloads {
+		preemptionVariant := classifyPreemptionVariant(ctx, candidateWl, hasHiearchicalAdvantage)
+		if preemptionVariant == Never {
+			continue
+		}
+		candidates = append(candidates,
+			&candidateElem{
+				wl:                candidateWl,
+				lca:               lca,
+				preemptionVariant: preemptionVariant,
+			})
+	}
+	return candidates
+}
+
+func collectCandidatesForHierarchicalReclaim(ctx *HierarchicalPreemptionCtx) ([]*candidateElem, []*candidateElem) {
+	hierarchyCandidates := []*candidateElem{}
+	priorityCandidates := []*candidateElem{}
+	if !ctx.Cq.HasParent() || ctx.Cq.Preemption.ReclaimWithinCohort == kueue.PreemptionPolicyNever {
+		return hierarchyCandidates, priorityCandidates
+	}
+	var previousSubtreeRoot *cache.CohortSnapshot
+	var candidateList *[]*candidateElem
+	var fits bool
+	hasHierarchicalAdvantage, remainingRequests := cache.QuantitiesFitInQuota(ctx.Cq, ctx.Requests)
+	for currentSubtreeRoot := range ctx.Cq.PathParentToRoot() {
+		if hasHierarchicalAdvantage {
+			candidateList = &hierarchyCandidates
+		} else {
+			candidateList = &priorityCandidates
+		}
+		collectCandidatesInSubtree(ctx, currentSubtreeRoot, currentSubtreeRoot, previousSubtreeRoot, hasHierarchicalAdvantage, candidateList)
+		fits, remainingRequests = cache.QuantitiesFitInQuota(currentSubtreeRoot, remainingRequests)
+		// Once we find a subtree sT that fits the requests, we will look for workloads that use quota
+		// of that subtree. The preemptor will have hierarchical advantage over all such workloads
+		// because it belongs to subtree sT. For that reason variable hasHierarchicalAdvantage
+		// remains true in subsequent iterations of the loop.
+		hasHierarchicalAdvantage = hasHierarchicalAdvantage || fits
+		previousSubtreeRoot = currentSubtreeRoot
+	}
+	return hierarchyCandidates, priorityCandidates
+}
+
+// visit the nodes in the hierarchy and collect the ones that exceed quota
+// avoid subtrees that are within quota and the skipped subtree
+func collectCandidatesInSubtree(ctx *HierarchicalPreemptionCtx, currentCohort *cache.CohortSnapshot, subtreeRoot *cache.CohortSnapshot, skipSubtree *cache.CohortSnapshot, hasHierarchicalAdvantage bool, result *[]*candidateElem) {
+	for _, childCohort := range currentCohort.ChildCohorts() {
+		// we already processed this subtree
+		if childCohort == skipSubtree {
+			continue
+		}
+		// don't look for candidates in subtrees that are not exceeding their quotas
+		if cache.IsWithinNominalInResources(childCohort, ctx.FrsNeedPreemption) {
+			continue
+		}
+		collectCandidatesInSubtree(ctx, childCohort, subtreeRoot, skipSubtree, hasHierarchicalAdvantage, result)
+	}
+	for _, childCq := range currentCohort.ChildCQs() {
+		if childCq == ctx.Cq {
+			continue
+		}
+		if !cache.IsWithinNominalInResources(childCq, ctx.FrsNeedPreemption) {
+			*result = append(*result, getCandidatesFromCQ(childCq, subtreeRoot, ctx, hasHierarchicalAdvantage)...)
+		}
+	}
+}
+
+// getNodeHeight calculates the distance to the furthest leaf
+func getNodeHeight(node *cache.CohortSnapshot) int {
+	maxHeight := min(node.ChildCount(), 1)
+	for _, childCohort := range node.ChildCohorts() {
+		maxHeight = max(maxHeight, getNodeHeight(childCohort)+1)
+	}
+	return maxHeight
+}
+
+// FindHeightOfLowestSubtreeThatFits returns height of a lowest subtree in the cohort
+// that fits additional val of resource fr. If no such subtree exists, it returns
+// height the whole cohort hierarchy. Note that height of a trivial subtree
+// with only one node is 0. It also returns if the returned subtree is smaller than the whole cohort tree.
+func FindHeightOfLowestSubtreeThatFits(c *cache.ClusterQueueSnapshot, fr resources.FlavorResource, val int64) (int, bool) {
+	if !(c.BorrowingWith(fr, val) && c.HasParent()) {
+		return 0, c.HasParent()
+	}
+	remaining := val - cache.LocalAvailable(c, fr)
+	for trackingNode := range c.PathParentToRoot() {
+		if !trackingNode.BorrowingWith(fr, remaining) {
+			return getNodeHeight(trackingNode), trackingNode.HasParent()
+		}
+		remaining -= cache.LocalAvailable(trackingNode, fr)
+	}
+	// no fit found
+	return getNodeHeight(c.Parent().Root()), false
+}
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index a3f5f35c5..a089a0e65 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -42,6 +42,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/resources"
 	"sigs.k8s.io/kueue/pkg/scheduler/flavorassigner"
+	"sigs.k8s.io/kueue/pkg/scheduler/preemption/classical"
 	"sigs.k8s.io/kueue/pkg/scheduler/preemption/fairsharing"
 	"sigs.k8s.io/kueue/pkg/util/priority"
 	"sigs.k8s.io/kueue/pkg/util/routine"
@@ -97,26 +98,6 @@ func (p *Preemptor) OverrideApply(f func(context.Context, *kueue.Workload, strin
 	p.applyPreemption = f
 }
 
-func candidatesOnlyFromQueue(candidates []*workload.Info, clusterQueue kueue.ClusterQueueReference) []*workload.Info {
-	result := make([]*workload.Info, 0, len(candidates))
-	for _, wi := range candidates {
-		if wi.ClusterQueue == clusterQueue {
-			result = append(result, wi)
-		}
-	}
-	return result
-}
-
-func candidatesFromCQOrUnderThreshold(candidates []*workload.Info, clusterQueue kueue.ClusterQueueReference, threshold int32) []*workload.Info {
-	result := make([]*workload.Info, 0, len(candidates))
-	for _, wi := range candidates {
-		if wi.ClusterQueue == clusterQueue || priority.Priority(wi.Obj) < threshold {
-			result = append(result, wi)
-		}
-	}
-	return result
-}
-
 type Target struct {
 	WorkloadInfo *workload.Info
 	Reason       string
@@ -142,65 +123,10 @@ func (p *Preemptor) GetTargets(log logr.Logger, wl workload.Info, assignment fla
 }
 
 func (p *Preemptor) getTargets(preemptionCtx *preemptionCtx) []*Target {
-	candidates := p.findCandidates(preemptionCtx.preemptor.Obj, preemptionCtx.preemptorCQ, preemptionCtx.frsNeedPreemption)
-	if len(candidates) == 0 {
-		return nil
-	}
-	sort.Slice(candidates, candidatesOrdering(candidates, preemptionCtx.preemptorCQ.Name, p.clock.Now()))
 	if p.enableFairSharing {
-		return fairPreemptions(preemptionCtx, candidates, p.fsStrategies)
-	}
-
-	sameQueueCandidates := candidatesOnlyFromQueue(candidates, preemptionCtx.preemptorCQ.Name)
-
-	// To avoid flapping, Kueue only allows preemption of workloads from the same
-	// queue if borrowing. Preemption of workloads from queues can happen only
-	// if not borrowing at the same time. Kueue prioritizes preemption of
-	// workloads from the other queues (that borrowed resources) first, before
-	// trying to preempt more own workloads and borrow at the same time.
-	if len(sameQueueCandidates) == len(candidates) {
-		// There is no possible preemption of workloads from other queues,
-		// so we'll try borrowing.
-		return minimalPreemptions(preemptionCtx, candidates, true, nil)
-	}
-
-	// 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, 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)
-		}
-		return minimalPreemptions(preemptionCtx, candidates, true, thresholdPrio)
-	}
-
-	// Only try preemptions in the cohort, without borrowing, if the target clusterqueue is still
-	// under nominal quota for all resources.
-	if queueUnderNominalInResourcesNeedingPreemption(preemptionCtx) {
-		if targets := minimalPreemptions(preemptionCtx, candidates, false, nil); len(targets) > 0 {
-			return targets
-		}
+		return p.fairPreemptions(preemptionCtx, p.fsStrategies)
 	}
-
-	// Final attempt. This time only candidates from the same queue, but
-	// with borrowing.
-	return minimalPreemptions(preemptionCtx, sameQueueCandidates, true, nil)
-}
-
-// canBorrowWithinCohort returns whether the behavior is enabled for the ClusterQueue and the threshold priority to use.
-func canBorrowWithinCohort(preemptionCtx *preemptionCtx) (bool, *int32) {
-	borrowWithinCohort := preemptionCtx.preemptorCQ.Preemption.BorrowWithinCohort
-	if borrowWithinCohort == nil || borrowWithinCohort.Policy == kueue.BorrowWithinCohortPolicyNever {
-		return false, nil
-	}
-	threshold := priority.Priority(preemptionCtx.preemptor.Obj)
-	if borrowWithinCohort.MaxPriorityThreshold != nil && *borrowWithinCohort.MaxPriorityThreshold < threshold {
-		threshold = *borrowWithinCohort.MaxPriorityThreshold + 1
-	}
-	return true, &threshold
+	return p.classicalPreemptions(preemptionCtx)
 }
 
 var HumanReadablePreemptionReasons = map[string]string{
@@ -264,66 +190,71 @@ func (p *Preemptor) applyPreemptionWithSSA(ctx context.Context, w *kueue.Workloa
 	return workload.ApplyAdmissionStatus(ctx, p.client, w, true, p.clock)
 }
 
-// minimalPreemptions implements a heuristic to find a minimal set of Workloads
+type preemptionAttemptOpts struct {
+	borrowing bool
+}
+
+// classicalPreemptions implements a heuristic to find a minimal set of Workloads
 // to preempt.
 // The heuristic first removes candidates, in the input order, while their
 // ClusterQueues are still borrowing resources and while the incoming Workload
 // doesn't fit in the quota.
 // Once the Workload fits, the heuristic tries to add Workloads back, in the
 // reverse order in which they were removed, while the incoming Workload still
-// fits.
-func minimalPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info, allowBorrowing bool, allowBorrowingBelowPriority *int32) []*Target {
-	if logV := preemptionCtx.log.V(5); logV.Enabled() {
-		logV.Info("Simulating preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "allowBorrowing", allowBorrowing, "allowBorrowingBelowPriority", allowBorrowingBelowPriority, "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
-	}
-	// Simulate removing all candidates from the ClusterQueue and cohort.
-	var targets []*Target
-	fits := false
-	for _, candWl := range candidates {
-		candCQ := preemptionCtx.snapshot.ClusterQueue(candWl.ClusterQueue)
-		reason := kueue.InClusterQueueReason
-		if preemptionCtx.preemptorCQ != candCQ {
-			if !cqIsBorrowing(candCQ, preemptionCtx.frsNeedPreemption) {
-				continue
-			}
-			reason = kueue.InCohortReclamationReason
-			if allowBorrowingBelowPriority != nil {
-				if priority.Priority(candWl.Obj) >= *allowBorrowingBelowPriority {
-					// We set allowBorrowing=false if there is a candidate with priority
-					// exceeding allowBorrowingBelowPriority added to targets.
-					//
-					// We need to be careful mutating allowBorrowing. We rely on the
-					// fact that once there is a candidate exceeding the priority added
-					// to targets, then at least one such candidate is present in the
-					// final set of targets (after the second phase of the function).
-					//
-					// This is true, because the candidates are ordered according
-					// to priorities (from lowest to highest, using candidatesOrdering),
-					// and the last added target is not removed in the second phase of
-					// the function.
-					allowBorrowing = false
-				} else {
-					reason = kueue.InCohortReclaimWhileBorrowingReason
-				}
+// fits
+func (p *Preemptor) classicalPreemptions(preemptionCtx *preemptionCtx) []*Target {
+	hierarchicalReclaimCtx := &classical.HierarchicalPreemptionCtx{
+		Wl:                preemptionCtx.preemptor.Obj,
+		Cq:                preemptionCtx.preemptorCQ,
+		FrsNeedPreemption: preemptionCtx.frsNeedPreemption,
+		Requests:          preemptionCtx.workloadUsage.Quota,
+		WorkloadOrdering:  p.workloadOrdering,
+	}
+	candidatesGenerator := classical.NewCandidateIterator(hierarchicalReclaimCtx, preemptionCtx.frsNeedPreemption, preemptionCtx.snapshot, p.clock, CandidatesOrdering)
+	var attemptPossibleOpts []preemptionAttemptOpts
+	borrowWithinCohortForbidden, _ := classical.IsBorrowingWithinCohortForbidden(preemptionCtx.preemptorCQ)
+	// We have three types of candidates:
+	// 1. Hierarchy candidates. Candidates over which the incoming workload has a
+	// 	  hierarchical advantage (it is closer to the quota used by the candidate).
+	//    We can preempt such candidates regardless of their priority.
+	// 2. Priority candidates. Candidates over which there is no hiearchical advantage
+	//    but the possibility to preempt is determined based on priorities.
+	// 	  We respect the BorrowWithinCohort configuration only for these candidates.
+	// 3. Same queue candidates.
+	// We can only preempt a priority candidate with priority > MaxPriorityThreshold
+	// if the target CQ is not borrowing (by the definition of the MaxPriorityThreshold).
+	// We sometimes need to consider both options allowBorrowing = true and false
+	// (because with false we have more candidates but cannot use borrowing).
+	// The order in which the options are considered is arbitrary and the condition
+	// in which we try allowBorrowing=false before true is to keep compatibility with
+	// previous versions.
+	switch {
+	case candidatesGenerator.NoCandidateFromOtherQueues || (borrowWithinCohortForbidden && !queueUnderNominalInResourcesNeedingPreemption(preemptionCtx)):
+		attemptPossibleOpts = []preemptionAttemptOpts{{true}}
+	case borrowWithinCohortForbidden && candidatesGenerator.NoCandidateForHierarchicalReclaim:
+		attemptPossibleOpts = []preemptionAttemptOpts{{false}, {true}}
+	default:
+		attemptPossibleOpts = []preemptionAttemptOpts{{true}, {false}}
+	}
+
+	for _, attemptOpts := range attemptPossibleOpts {
+		var targets []*Target
+		candidatesGenerator.Reset()
+		for candidate, reason := candidatesGenerator.Next(attemptOpts.borrowing); candidate != nil; candidate, reason = candidatesGenerator.Next(attemptOpts.borrowing) {
+			preemptionCtx.snapshot.RemoveWorkload(candidate)
+			targets = append(targets, &Target{
+				WorkloadInfo: candidate,
+				Reason:       reason,
+			})
+			if workloadFits(preemptionCtx, attemptOpts.borrowing) {
+				targets = fillBackWorkloads(preemptionCtx, targets, attemptOpts.borrowing)
+				restoreSnapshot(preemptionCtx.snapshot, targets)
+				return targets
 			}
 		}
-		preemptionCtx.snapshot.RemoveWorkload(candWl)
-		targets = append(targets, &Target{
-			WorkloadInfo: candWl,
-			Reason:       reason,
-		})
-		if workloadFits(preemptionCtx, allowBorrowing) {
-			fits = true
-			break
-		}
-	}
-	if !fits {
 		restoreSnapshot(preemptionCtx.snapshot, targets)
-		return nil
 	}
-	targets = fillBackWorkloads(preemptionCtx, targets, allowBorrowing)
-	restoreSnapshot(preemptionCtx.snapshot, targets)
-	return targets
+	return nil
 }
 
 func fillBackWorkloads(preemptionCtx *preemptionCtx, targets []*Target, allowBorrowing bool) []*Target {
@@ -439,7 +370,12 @@ func runSecondFsStrategy(retryCandidates []*workload.Info, preemptionCtx *preemp
 	return false, targets
 }
 
-func fairPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info, strategies []fairsharing.Strategy) []*Target {
+func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, strategies []fairsharing.Strategy) []*Target {
+	candidates := p.findCandidates(preemptionCtx.preemptor.Obj, preemptionCtx.preemptorCQ, preemptionCtx.frsNeedPreemption)
+	if len(candidates) == 0 {
+		return nil
+	}
+	sort.Slice(candidates, CandidatesOrdering(candidates, preemptionCtx.preemptorCQ.Name, p.clock.Now()))
 	if logV := preemptionCtx.log.V(5); logV.Enabled() {
 		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
 	}
@@ -495,7 +431,7 @@ func (p *Preemptor) findCandidates(wl *kueue.Workload, cq *cache.ClusterQueueSna
 				continue
 			}
 
-			if !workloadUsesResources(candidateWl, frsNeedPreemption) {
+			if !classical.WorkloadUsesResources(candidateWl, frsNeedPreemption) {
 				continue
 			}
 			candidates = append(candidates, candidateWl)
@@ -513,7 +449,7 @@ func (p *Preemptor) findCandidates(wl *kueue.Workload, cq *cache.ClusterQueueSna
 				if onlyLowerPriority && priority.Priority(candidateWl.Obj) >= priority.Priority(wl) {
 					continue
 				}
-				if !workloadUsesResources(candidateWl, frsNeedPreemption) {
+				if !classical.WorkloadUsesResources(candidateWl, frsNeedPreemption) {
 					continue
 				}
 				candidates = append(candidates, candidateWl)
@@ -535,17 +471,6 @@ func cqIsBorrowing(cq *cache.ClusterQueueSnapshot, frsNeedPreemption sets.Set[re
 	return false
 }
 
-func workloadUsesResources(wl *workload.Info, frsNeedPreemption sets.Set[resources.FlavorResource]) bool {
-	for _, ps := range wl.TotalRequests {
-		for res, flv := range ps.Flavors {
-			if frsNeedPreemption.Has(resources.FlavorResource{Flavor: flv, Resource: res}) {
-				return true
-			}
-		}
-	}
-	return false
-}
-
 // workloadFits determines if the workload requests would fit given the
 // requestable resources and simulated usage of the ClusterQueue and its cohort,
 // if it belongs to one.
@@ -588,7 +513,7 @@ func queueUnderNominalInResourcesNeedingPreemption(preemptionCtx *preemptionCtx)
 // same ClusterQueue as the preemptor.
 // 2. Workloads with lower priority first.
 // 3. Workloads admitted more recently first.
-func candidatesOrdering(candidates []*workload.Info, cq kueue.ClusterQueueReference, now time.Time) func(int, int) bool {
+func CandidatesOrdering(candidates []*workload.Info, cq kueue.ClusterQueueReference, now time.Time) func(int, int) bool {
 	return func(i, j int) bool {
 		a := candidates[i]
 		b := candidates[j]
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 8e24abe4e..05fda284a 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -402,7 +402,7 @@ func quotaResourcesToReserve(e *entry, cq *cache.ClusterQueueSnapshot) resources
 	reservedUsage := make(resources.FlavorResourceQuantities)
 	for fr, usage := range e.assignment.Usage.Quota {
 		cqQuota := cq.QuotaFor(fr)
-		if e.assignment.Borrowing {
+		if e.assignment.Borrowing > 0 {
 			if cqQuota.BorrowingLimit == nil {
 				reservedUsage[fr] = usage
 			} else {
@@ -580,7 +580,7 @@ func (e entryOrdering) Less(i, j int) bool {
 	aBorrows := a.assignment.Borrows()
 	bBorrows := b.assignment.Borrows()
 	if aBorrows != bBorrows {
-		return !aBorrows
+		return aBorrows < bBorrows
 	}
 
 	// 2. Higher priority first if not disabled.
@@ -651,8 +651,7 @@ func (s *Scheduler) requeueAndUpdate(ctx context.Context, e entry) {
 	log.V(2).Info("Workload re-queued", "workload", klog.KObj(e.Obj), "clusterQueue", klog.KRef("", string(e.ClusterQueue)), "queue", klog.KRef(e.Obj.Namespace, e.Obj.Spec.QueueName), "requeueReason", e.requeueReason, "added", added, "status", e.status)
 
 	if e.status == notNominated || e.status == skipped {
-		patch := workload.BaseSSAWorkload(e.Obj)
-		workload.AdmissionStatusPatch(e.Obj, patch, true)
+		patch := workload.PrepareWorkloadPatch(e.Obj, true, s.clock)
 		reservationIsChanged := workload.UnsetQuotaReservationWithCondition(patch, "Pending", e.inadmissibleMsg, s.clock.Now())
 		resourceRequestsIsChanged := workload.PropagateResourceRequests(patch, &e.Info)
 		if reservationIsChanged || resourceRequestsIsChanged {
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index c368941f9..f66e871d6 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -709,10 +709,15 @@ func AdmissionChecksStatusPatch(w *kueue.Workload, wlCopy *kueue.Workload, c clo
 // If strict is true, resourceVersion will be part of the patch, make this call fail if Workload
 // was changed.
 func ApplyAdmissionStatus(ctx context.Context, c client.Client, w *kueue.Workload, strict bool, clk clock.Clock) error {
+	wlCopy := PrepareWorkloadPatch(w, strict, clk)
+	return ApplyAdmissionStatusPatch(ctx, c, wlCopy)
+}
+
+func PrepareWorkloadPatch(w *kueue.Workload, strict bool, clk clock.Clock) *kueue.Workload {
 	wlCopy := BaseSSAWorkload(w)
 	AdmissionStatusPatch(w, wlCopy, strict)
 	AdmissionChecksStatusPatch(w, wlCopy, clk)
-	return ApplyAdmissionStatusPatch(ctx, c, wlCopy)
+	return wlCopy
 }
 
 // ApplyAdmissionStatusPatch applies the patch of admission related status fields of a workload with SSA.



Test patch:
diff --git a/pkg/scheduler/flavorassigner/flavorassigner_test.go b/pkg/scheduler/flavorassigner/flavorassigner_test.go
index fc333b18f..638cf3242 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner_test.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner_test.go
@@ -26,6 +26,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -410,7 +411,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Borrowing: true,
+				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "two", Resource: corev1.ResourceCPU}:    3_000,
 					{Flavor: "two", Resource: corev1.ResourceMemory}: 10 * utiltesting.Mi,
@@ -805,7 +806,7 @@ func TestAssignFlavors(t *testing.T) {
 						Count: 1,
 					},
 				},
-				Borrowing: true,
+				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "default", Resource: corev1.ResourceCPU}:    10_000,
 					{Flavor: "default", Resource: corev1.ResourceMemory}: 5 * utiltesting.Gi,
@@ -892,7 +893,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Borrowing: true,
+				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 2_000,
 				}},
@@ -974,7 +975,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Borrowing: true,
+				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}: 2_000,
 				}},
@@ -1340,7 +1341,7 @@ func TestAssignFlavors(t *testing.T) {
 				Obj(),
 			wantRepMode: Fit,
 			wantAssignment: Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 				PodSets: []PodSetAssignment{{
 					Name: kueue.DefaultPodSetName,
 					Flavors: ResourceAssignment{
@@ -1441,7 +1442,7 @@ func TestAssignFlavors(t *testing.T) {
 				Obj(),
 			wantRepMode: Fit,
 			wantAssignment: Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 				PodSets: []PodSetAssignment{{
 					Name: kueue.DefaultPodSetName,
 					Flavors: ResourceAssignment{
@@ -1498,7 +1499,7 @@ func TestAssignFlavors(t *testing.T) {
 			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 				PodSets: []PodSetAssignment{{
 					Name: kueue.DefaultPodSetName,
 					Flavors: ResourceAssignment{
@@ -1555,7 +1556,7 @@ func TestAssignFlavors(t *testing.T) {
 			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 				PodSets: []PodSetAssignment{{
 					Name: kueue.DefaultPodSetName,
 					Flavors: ResourceAssignment{
@@ -1768,7 +1769,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Borrowing: true,
+				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}:  9_000,
 					{Flavor: "one", Resource: corev1.ResourcePods}: 1,
@@ -1818,7 +1819,7 @@ func TestAssignFlavors(t *testing.T) {
 					},
 					Count: 1,
 				}},
-				Borrowing: true,
+				Borrowing: 1,
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
 					{Flavor: "one", Resource: corev1.ResourceCPU}:  9_000,
 					{Flavor: "one", Resource: corev1.ResourcePods}: 1,
@@ -1856,7 +1857,7 @@ func TestAssignFlavors(t *testing.T) {
 			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 				PodSets: []PodSetAssignment{{
 					Name: kueue.DefaultPodSetName,
 					Flavors: ResourceAssignment{
@@ -2322,3 +2323,135 @@ func TestLastAssignmentOutdated(t *testing.T) {
 		})
 	}
 }
+
+// We have 3 flavors: one, two, three and a 3-level cohort hierarchy where
+// each cohort get 4 units of CPU in a different flavor.
+func TestHierarchical(t *testing.T) {
+	type rfMap = map[corev1.ResourceName]kueue.ResourceFlavorReference
+	cases := map[string]struct {
+		workloadRequests       *utiltesting.PodSetWrapper
+		testClusterQueueUsage  resources.FlavorResourceQuantities
+		otherClusterQueueUsage resources.FlavorResourceQuantities
+		flavorFungibility      *kueue.FlavorFungibility
+		wantMode               FlavorAssignmentMode
+		wantAssigment          rfMap
+	}{
+		"Select the top flavor": {
+			workloadRequests: utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "4"),
+			testClusterQueueUsage: resources.FlavorResourceQuantities{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: 4_000,
+			},
+			otherClusterQueueUsage: resources.FlavorResourceQuantities{
+				{Flavor: "two", Resource: corev1.ResourceCPU}: 4_000,
+			},
+			wantMode:      Fit,
+			wantAssigment: rfMap{corev1.ResourceCPU: "three"},
+		},
+		"Select the first flavor which fits": {
+			workloadRequests: utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "4"),
+			testClusterQueueUsage: resources.FlavorResourceQuantities{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: 4_000,
+			},
+			wantMode:      Fit,
+			wantAssigment: rfMap{corev1.ResourceCPU: "two"},
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			ctx, _ := utiltesting.ContextWithLog(t)
+			resourceFlavors := map[kueue.ResourceFlavorReference]*kueue.ResourceFlavor{
+				"one":   utiltesting.MakeResourceFlavor("one").Obj(),
+				"two":   utiltesting.MakeResourceFlavor("two").Obj(),
+				"three": utiltesting.MakeResourceFlavor("three").Obj(),
+			}
+			cohorts := []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("three").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("three").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("two").
+					Parent("three").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("one").
+					Parent("two").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+			}
+			testCq := utiltesting.MakeClusterQueue("test-clusterqueue").
+				Cohort("one").
+				Preemption(kueue.ClusterQueuePreemption{
+					WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+					ReclaimWithinCohort: kueue.PreemptionPolicyLowerPriority,
+				}).
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").Resource(corev1.ResourceCPU, "0").FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").Resource(corev1.ResourceCPU, "0").FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("three").Resource(corev1.ResourceCPU, "0").FlavorQuotas,
+				).
+				FlavorFungibility(kueue.FlavorFungibility{
+					WhenCanPreempt: kueue.TryNextFlavor,
+				}).ClusterQueue
+			otherCq := utiltesting.MakeClusterQueue("other-clusterqueue").
+				Cohort("two").ResourceGroup(
+				utiltesting.MakeFlavorQuotas("one").Resource(corev1.ResourceCPU, "0").FlavorQuotas,
+				utiltesting.MakeFlavorQuotas("two").Resource(corev1.ResourceCPU, "0").FlavorQuotas,
+				utiltesting.MakeFlavorQuotas("three").Resource(corev1.ResourceCPU, "0").FlavorQuotas,
+			).ClusterQueue
+
+			wlInfo := workload.NewInfo(&kueue.Workload{
+				Spec: kueue.WorkloadSpec{
+					PodSets: []kueue.PodSet{
+						tc.workloadRequests.PodSet,
+					},
+				},
+			})
+
+			if tc.flavorFungibility != nil {
+				testCq.Spec.FlavorFungibility = tc.flavorFungibility
+			}
+			cache := cache.New(utiltesting.NewFakeClient())
+			for _, cohort := range cohorts {
+				if err := cache.AddOrUpdateCohort(cohort); err != nil {
+					t.Fatalf("Couldn't add Cohort to cache: %v", err)
+				}
+			}
+			if err := cache.AddClusterQueue(ctx, &testCq); err != nil {
+				t.Fatalf("Failed to add CQ to cache")
+			}
+			if err := cache.AddClusterQueue(ctx, &otherCq); err != nil {
+				t.Fatalf("Failed to add CQ to cache")
+			}
+			for _, rf := range resourceFlavors {
+				cache.AddOrUpdateResourceFlavor(rf)
+			}
+
+			snapshot, err := cache.Snapshot(ctx)
+			if err != nil {
+				t.Fatalf("unexpected error while building snapshot: %v", err)
+			}
+			otherClusterQueue := snapshot.ClusterQueue("other-clusterqueue")
+			otherClusterQueue.AddUsage(workload.Usage{Quota: tc.otherClusterQueueUsage})
+
+			testClusterQueue := snapshot.ClusterQueue("test-clusterqueue")
+			testClusterQueue.AddUsage(workload.Usage{Quota: tc.testClusterQueueUsage})
+
+			flvAssigner := New(wlInfo, testClusterQueue, resourceFlavors, false, &testOracle{})
+			log := testr.NewWithOptions(t, testr.Options{Verbosity: 2})
+			assignment := flvAssigner.Assign(log, nil)
+			if gotRepMode := assignment.RepresentativeMode(); gotRepMode != tc.wantMode {
+				t.Errorf("Unexpected RepresentativeMode. got %s, want %s", gotRepMode, tc.wantMode)
+			}
+			if len(assignment.PodSets[0].Flavors) != len(tc.wantAssigment) {
+				t.Errorf("Wrong number of flavors. got %d, want %d", len(assignment.PodSets[0].Flavors), len(tc.wantAssigment))
+			}
+			for resourceName, wantFlavor := range tc.wantAssigment {
+				if gotFlavor := assignment.PodSets[0].Flavors[resourceName].Name; gotFlavor != wantFlavor {
+					t.Errorf("Unexpected flavor. got %s, want %s", gotFlavor, wantFlavor)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index 119f81ab1..06dab3a74 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -2729,7 +2729,6 @@ func TestFairPreemptions(t *testing.T) {
 func targetKeyReason(key, reason string) string {
 	return fmt.Sprintf("%s:%s", key, reason)
 }
-
 func TestCandidatesOrdering(t *testing.T) {
 	now := time.Now()
 	candidates := []*workload.Info{
@@ -2764,7 +2763,7 @@ func TestCandidatesOrdering(t *testing.T) {
 			ReserveQuotaAt(utiltesting.MakeAdmission("self").Obj(), now.Add(time.Second)).
 			Obj()),
 	}
-	sort.Slice(candidates, candidatesOrdering(candidates, "self", now))
+	sort.Slice(candidates, CandidatesOrdering(candidates, "self", now))
 	gotNames := make([]string, len(candidates))
 	for i, c := range candidates {
 		gotNames[i] = workload.Key(c.Obj)
@@ -2816,3 +2815,1207 @@ func TestPreemptionMessage(t *testing.T) {
 		}
 	}
 }
+
+func TestHierarchicalPreemptions(t *testing.T) {
+	now := time.Now()
+	flavors := []*kueue.ResourceFlavor{
+		utiltesting.MakeResourceFlavor("default").Obj(),
+	}
+	cases := map[string]struct {
+		clusterQueues []*kueue.ClusterQueue
+		cohorts       []*kueuealpha.Cohort
+		admitted      []kueue.Workload
+		incoming      *kueue.Workload
+		targetCQ      kueue.ClusterQueueReference
+		assignment    flavorassigner.Assignment
+		wantPreempted sets.Set[string]
+	}{
+		//
+		//            R
+		//      /      |
+		//   C(2) q_borrowing(0)
+		//  /
+		// q
+		"preempt with hierarchical advantage": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted2", "").
+					Priority(0).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "2").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(targetKeyReason("/admitted2", kueue.InCohortReclamationReason)),
+		},
+		//
+		//             R
+		//      /      |         \
+		//   C(2) q_borrowing(0)  q_nominal(2)
+		//  /
+		// q
+		"avoid queues within nominal quota": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q_nominal").
+					Cohort("r").ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+					Resource(corev1.ResourceCPU, "2").
+					Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted1", "").
+					Priority(-10).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_nominal").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted2", "").
+					Priority(0).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "2").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(targetKeyReason("/admitted2", kueue.InCohortReclamationReason)),
+		},
+		//
+		//            R(0)
+		//      /      |
+		//   C(2) q_borrowing(0)
+		//  /
+		// q(0)
+		"preempt multiple with hierarchical advantage": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted1", "").
+					Priority(1).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted2", "").
+					Priority(2).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "2").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted1", kueue.InCohortReclamationReason),
+				targetKeyReason("/admitted2", kueue.InCohortReclamationReason)),
+		},
+		//
+		//        R(0)
+		//      /
+		//   C(3)
+		//  /   \
+		// q(0) q_same_cohort(0)
+		"preempt in cohort and own CQ": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "3").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+						BorrowWithinCohort: &kueue.BorrowWithinCohort{
+							Policy:               kueue.BorrowWithinCohortPolicyLowerPriority,
+							MaxPriorityThreshold: ptr.To[int32](0),
+						},
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_not_preemptible", "").
+					Priority(1).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_preemptible", "").
+					Priority(0).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_own_queue", "").
+					Priority(-1).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(1).
+				Request(corev1.ResourceCPU, "2").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_preemptible", kueue.InCohortReclaimWhileBorrowingReason),
+				targetKeyReason("/admitted_own_queue", kueue.InClusterQueueReason)),
+		},
+		//
+		//            R(0)
+		//      /      |
+		//   C(2) q_borrowing(0)
+		//  /
+		// q(0)
+		"prefer to preempt hierarchical candidate": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing", "").
+					Priority(1).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_same_queue", "").
+					Priority(-2).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "1").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_borrowing", kueue.InCohortReclamationReason)),
+		},
+		//
+		//           R(0)
+		//      /      |
+		//   C(2)   q_nominal(2)
+		//  /   \
+		// q(0) q_same_cohort(0)
+		"forced to preempt priority candidate": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+						BorrowWithinCohort: &kueue.BorrowWithinCohort{
+							Policy:               kueue.BorrowWithinCohortPolicyLowerPriority,
+							MaxPriorityThreshold: ptr.To[int32](0),
+						},
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_nominal").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_nominal", "").
+					Priority(-10).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_nominal").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_same_cohort", "").
+					Priority(-1).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "2").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_same_cohort", kueue.InCohortReclaimWhileBorrowingReason)),
+		},
+		//
+		//           R(0)
+		//      /      |
+		//   C(2)   q_borrowing(0)
+		//  /    \
+		// q(4)  q_same_cohort(0)
+		//
+		"incoming workload fits in CQ nominal quota": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "4").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing", "").
+					Priority(10).
+					Request(corev1.ResourceCPU, "3").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "3").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_same_cohort", "").
+					Priority(10).
+					Request(corev1.ResourceCPU, "3").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "3").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "4").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_borrowing", kueue.InCohortReclamationReason),
+				targetKeyReason("/admitted_same_cohort", kueue.InCohortReclamationReason)),
+		},
+		//
+		//           R(1)
+		//      /      |
+		//   C(4)   q_borrowing(0)
+		//  /    \
+		// q(0)  q_same_cohort(0)
+		//
+		"preempt hierarchical and priority candidates": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+					Resource(corev1.ResourceCPU, "1").
+					Obj()).Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyLowerPriority,
+						BorrowWithinCohort: &kueue.BorrowWithinCohort{
+							Policy:               kueue.BorrowWithinCohortPolicyLowerPriority,
+							MaxPriorityThreshold: ptr.To[int32](0),
+						},
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing", "").
+					Priority(-1).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_same_cohort_preemptible", "").
+					Priority(-1).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_not_preemptible", "").
+					Priority(1).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "3").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_borrowing", kueue.InCohortReclamationReason),
+				targetKeyReason("/admitted_same_cohort_preemptible", kueue.InCohortReclaimWhileBorrowingReason)),
+		},
+		//
+		//           R(1)
+		//      /      |
+		//   C(4)   q_borrowing(0)
+		//  /    \
+		// q(0)  q_same_cohort(0)
+		//
+		"preempt hierarchical candidates and inside CQ": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+					Resource(corev1.ResourceCPU, "1").
+					Obj()).Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyLowerPriority,
+						BorrowWithinCohort: &kueue.BorrowWithinCohort{
+							Policy:               kueue.BorrowWithinCohortPolicyLowerPriority,
+							MaxPriorityThreshold: ptr.To[int32](0),
+						},
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing", "").
+					Priority(-1).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_same_queue_preemptible", "").
+					Priority(-1).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_not_preemptible", "").
+					Priority(1).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "3").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_borrowing", kueue.InCohortReclamationReason),
+				targetKeyReason("/admitted_same_queue_preemptible", kueue.InClusterQueueReason)),
+		},
+		//
+		//            R(0)
+		//      /      |          \
+		//   C(3) q_borrowing(0)  q_nominal(2)
+		//  /
+		// q(0)
+		"reclaim nominal quota from lowest priority workload, excluding non-borrowing": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "3").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_nominal").
+					Cohort("r").ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+					Resource(corev1.ResourceCPU, "2").
+					Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing_prio_8", "").
+					Priority(8).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_prio_9", "").
+					Priority(9).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_prio_10", "").
+					Priority(9).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_nominal", "").
+					Priority(-2).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_nominal").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "1").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_borrowing_prio_8", kueue.InCohortReclamationReason)),
+		},
+		//
+		//                  R
+		//            /            \
+		//      C(2)                   C_other(2)
+		//     /    \                     |
+		//    q(0)  q_same_cohort(0)   q_other(0)
+		"infeasible preemption all available workloads in pruned subtrees": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c_other").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q_other").
+					Cohort("c_other").ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+					Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_other_1", "").
+					Priority(-10).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_other").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_other_2", "").
+					Priority(-10).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_other").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_same_cohort", "").
+					Priority(0).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "2").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New[string](),
+		},
+		//
+		//          R(3CPU, 0Gi)
+		//      /      |
+		//   C(4CPU,4Gi) q_borrowing(0)
+		//  /    \
+		// q(0)   q_same_cohort(0)
+		"hiearchical preemption with multiple resources": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "3").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Resource(corev1.ResourceMemory, "4Gi").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+					Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing", "").
+					Priority(0).
+					Request(corev1.ResourceCPU, "3").
+					Request(corev1.ResourceMemory, "1Gi").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "3").
+						Assignment(corev1.ResourceMemory, "default", "1Gi").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_same_cohort", "").
+					Priority(-2).
+					Request(corev1.ResourceCPU, "1").
+					Request(corev1.ResourceMemory, "3Gi").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "1").
+						Assignment(corev1.ResourceMemory, "default", "3Gi").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(-2).
+				Request(corev1.ResourceCPU, "2").
+				Request(corev1.ResourceMemory, "1Gi").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+				corev1.ResourceMemory: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(targetKeyReason("/admitted_borrowing", kueue.InCohortReclamationReason)),
+		},
+		//
+		//           R(0)
+		//      /      |
+		//   C(2)   q_borrowing(0)
+		//  /    \
+		// q(0)  q_same_cohort(0)
+		//
+		"prefer to preempt evicted workloads": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+						BorrowWithinCohort: &kueue.BorrowWithinCohort{
+							Policy:               kueue.BorrowWithinCohortPolicyLowerPriority,
+							MaxPriorityThreshold: ptr.To[int32](0),
+						},
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_same_cohort").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing", "").
+					Priority(-10).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("evicted_same_cohort", "").
+					Priority(-1).
+					Request(corev1.ResourceCPU, "1").
+					ReserveQuota(utiltesting.MakeAdmission("q_same_cohort").
+						Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					SetOrReplaceCondition(metav1.Condition{
+						Type:               kueue.WorkloadEvicted,
+						Status:             metav1.ConditionTrue,
+						LastTransitionTime: metav1.NewTime(now),
+					}).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(0).
+				Request(corev1.ResourceCPU, "1").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New[string](),
+		},
+		//
+		//           R(0)
+		//      /      |
+		//   C(2)   q_borrowing(0)
+		//  /
+		// q(3, lending limit 2)
+		//
+		"respect lending limits": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "2").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q").
+					Cohort("c").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "3", "", "2").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+				utiltesting.MakeClusterQueue("q_borrowing").
+					Cohort("r").ResourceGroup(
+					*utiltesting.MakeFlavorQuotas("default").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing", "").
+					Priority(0).
+					Request(corev1.ResourceCPU, "4").
+					ReserveQuota(utiltesting.MakeAdmission("q_borrowing").
+						Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(-2).
+				Request(corev1.ResourceCPU, "5").
+				Obj(),
+			targetCQ: "q",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_borrowing", kueue.InCohortReclamationReason)),
+		},
+		//                                r
+		//                             /      \
+		//                          c11        c12
+		//                       /   |   \       \
+		//                    c21   c22    c23    q1
+		//                  /  |     |     |
+		//                c31  c32   q3    q2
+		//              /      |
+		//            q5       q4
+		//	quotas:
+		//	4: c11, c12, c21, c22, c23, c32, c31
+		//	0: q1, q3, q4, q5
+		"reclaim in complex hierarchy": {
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("r").Obj(),
+				utiltesting.MakeCohort("c11").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c12").
+					Parent("r").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c21").
+					Parent("c11").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c22").
+					Parent("c11").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c23").
+					Parent("c11").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c31").
+					Parent("c21").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+				utiltesting.MakeCohort("c32").
+					Parent("c21").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "4").
+						Obj()).Obj(),
+			},
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("q1").
+					Cohort("c12").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					Obj(),
+				utiltesting.MakeClusterQueue("q2").
+					Cohort("c23").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					Obj(),
+				utiltesting.MakeClusterQueue("q3").
+					Cohort("c22").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					Obj(),
+				utiltesting.MakeClusterQueue("q4").
+					Cohort("c32").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					Obj(),
+				utiltesting.MakeClusterQueue("q5").
+					Cohort("c31").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					Obj(),
+			},
+			admitted: []kueue.Workload{
+				*utiltesting.MakeWorkload("admitted_borrowing_1", "").
+					Priority(-6).
+					Request(corev1.ResourceCPU, "4").
+					ReserveQuota(utiltesting.MakeAdmission("q1").
+						Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_2", "").
+					Priority(-5).
+					Request(corev1.ResourceCPU, "4").
+					ReserveQuota(utiltesting.MakeAdmission("q1").
+						Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_3", "").
+					Priority(-9).
+					Request(corev1.ResourceCPU, "4").
+					ReserveQuota(utiltesting.MakeAdmission("q2").
+						Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_4", "").
+					Priority(-10).
+					Request(corev1.ResourceCPU, "4").
+					ReserveQuota(utiltesting.MakeAdmission("q2").
+						Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_5", "").
+					Priority(-4).
+					Request(corev1.ResourceCPU, "4").
+					ReserveQuota(utiltesting.MakeAdmission("q3").
+						Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_6", "").
+					Priority(-3).
+					Request(corev1.ResourceCPU, "3").
+					ReserveQuota(utiltesting.MakeAdmission("q3").
+						Assignment(corev1.ResourceCPU, "default", "3").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_7", "").
+					Priority(4).
+					Request(corev1.ResourceCPU, "2").
+					ReserveQuota(utiltesting.MakeAdmission("q4").
+						Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("admitted_borrowing_8", "").
+					Priority(2).
+					Request(corev1.ResourceCPU, "3").
+					ReserveQuota(utiltesting.MakeAdmission("q4").
+						Assignment(corev1.ResourceCPU, "default", "3").Obj()).
+					Obj(),
+			},
+			incoming: utiltesting.MakeWorkload("incoming", "").
+				Priority(-2).
+				Request(corev1.ResourceCPU, "7").
+				Obj(),
+			targetCQ: "q5",
+			assignment: singlePodSetAssignment(flavorassigner.ResourceAssignment{
+				corev1.ResourceCPU: &flavorassigner.FlavorAssignment{
+					Name: "default",
+					Mode: flavorassigner.Preempt,
+				},
+			}),
+			// only one of workloads from q2 will be preempted because
+			// after preempting the first one, the usage of cohort
+			// c23 will be back within nominal quota
+			wantPreempted: sets.New(
+				targetKeyReason("/admitted_borrowing_1", kueue.InCohortReclamationReason),
+				targetKeyReason("/admitted_borrowing_4", kueue.InCohortReclamationReason)),
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			ctx, log := utiltesting.ContextWithLog(t)
+			cl := utiltesting.NewClientBuilder().
+				WithLists(&kueue.WorkloadList{Items: tc.admitted}).
+				Build()
+
+			cqCache := cache.New(cl)
+			for _, flv := range flavors {
+				cqCache.AddOrUpdateResourceFlavor(flv)
+			}
+			for _, cq := range tc.clusterQueues {
+				if err := cqCache.AddClusterQueue(ctx, cq); err != nil {
+					t.Fatalf("Couldn't add ClusterQueue to cache: %v", err)
+				}
+			}
+			for _, cohort := range tc.cohorts {
+				if err := cqCache.AddOrUpdateCohort(cohort); err != nil {
+					t.Fatalf("Couldn't add Cohort to cache: %v", err)
+				}
+			}
+
+			var lock sync.Mutex
+			gotPreempted := sets.New[string]()
+			broadcaster := record.NewBroadcaster()
+			scheme := runtime.NewScheme()
+			if err := kueue.AddToScheme(scheme); err != nil {
+				t.Fatalf("Failed adding kueue scheme: %v", err)
+			}
+			recorder := broadcaster.NewRecorder(scheme, corev1.EventSource{Component: constants.AdmissionName})
+			preemptor := New(cl, workload.Ordering{}, recorder, config.FairSharing{}, clocktesting.NewFakeClock(now))
+			preemptor.applyPreemption = func(ctx context.Context, w *kueue.Workload, reason, _ string) error {
+				lock.Lock()
+				gotPreempted.Insert(targetKeyReason(workload.Key(w), reason))
+				lock.Unlock()
+				return nil
+			}
+			beforeSnapshot, err := cqCache.Snapshot(ctx)
+			if err != nil {
+				t.Fatalf("unexpected error while building snapshot: %v", err)
+			}
+			// make a working copy of the snapshotWorkingCopy than preemption can temporarily modify
+			snapshotWorkingCopy, err := cqCache.Snapshot(ctx)
+			if err != nil {
+				t.Fatalf("unexpected error while building snapshot: %v", err)
+			}
+			wlInfo := workload.NewInfo(tc.incoming)
+			wlInfo.ClusterQueue = tc.targetCQ
+			targets := preemptor.GetTargets(log, *wlInfo, tc.assignment, snapshotWorkingCopy)
+			_, err = preemptor.IssuePreemptions(ctx, wlInfo, targets)
+			if err != nil {
+				t.Fatalf("Failed doing preemption")
+			}
+			if diff := cmp.Diff(tc.wantPreempted, gotPreempted, cmpopts.EquateEmpty()); diff != "" {
+				t.Errorf("Issued preemptions (-want,+got):\n%s", diff)
+			}
+			if diff := cmp.Diff(beforeSnapshot, snapshotWorkingCopy, snapCmpOpts); diff != "" {
+				t.Errorf("Snapshot was modified (-initial,+end):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index be9486c34..e1114d496 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -3043,6 +3043,69 @@ func TestSchedule(t *testing.T) {
 				"eng-alpha/a1-admitted": *utiltesting.MakeAdmission("ClusterQueueA").Assignment("gpu", "on-demand", "1").Obj(),
 			},
 		},
+		//
+		// Denote Cohorts in UPPERCASE, and ClusterQueues in lowercase.
+		// quota is at GUARANTEED cohort
+		//
+		//             ROOT
+		//          /        \
+		//    GUARANTEED      best-effort
+		//   /
+		// guaranteed
+		//
+		"in a hierarchical cohort, workload borrowing less is scheduled first": {
+			cohorts: []kueuealpha.Cohort{
+				utiltesting.MakeCohort("root").Cohort,
+				utiltesting.MakeCohort("guaranteed").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("default").
+							Resource(corev1.ResourceCPU, "4").Obj(),
+					).Parent("root").Cohort,
+			},
+			additionalClusterQueues: []kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("guaranteed").
+					Cohort("guaranteed").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("default").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+				utiltesting.MakeClusterQueue("best-effort").
+					Cohort("root").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("default").
+							Resource(corev1.ResourceCPU, "0").Obj(),
+					).
+					ClusterQueue,
+			},
+			additionalLocalQueues: []kueue.LocalQueue{
+				*utiltesting.MakeLocalQueue("lq-guaranteed", "eng-alpha").ClusterQueue("guaranteed").Obj(),
+				*utiltesting.MakeLocalQueue("lq-best-effort", "eng-alpha").ClusterQueue("best-effort").Obj(),
+			},
+			workloads: []kueue.Workload{
+				utiltesting.MakeWorkload("guaranteed", "eng-alpha").
+					Queue("lq-guaranteed").
+					Priority(0).
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "4").
+						PodSet).
+					Workload,
+				utiltesting.MakeWorkload("best-effort", "eng-alpha").
+					Queue("lq-best-effort").
+					Priority(3).
+					PodSets(utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "4").
+						PodSet).
+					Workload,
+			},
+			wantAssignments: map[string]kueue.Admission{
+				"eng-alpha/guaranteed": *utiltesting.MakeAdmission("guaranteed", "one").Assignment(corev1.ResourceCPU, "default", "4").Obj(),
+			},
+			wantScheduled: []string{"eng-alpha/guaranteed"},
+			wantLeft: map[kueue.ClusterQueueReference][]string{
+				"best-effort": {"eng-alpha/best-effort"},
+			},
+		},
 	}
 
 	for name, tc := range cases {
@@ -3219,7 +3282,7 @@ func TestEntryOrdering(t *testing.T) {
 				}},
 			},
 			assignment: flavorassigner.Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 			},
 		},
 		{
@@ -3248,7 +3311,7 @@ func TestEntryOrdering(t *testing.T) {
 				}},
 			},
 			assignment: flavorassigner.Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 			},
 		},
 		{
@@ -3269,7 +3332,7 @@ func TestEntryOrdering(t *testing.T) {
 				}},
 			},
 			assignment: flavorassigner.Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 			},
 		},
 		{
@@ -3292,7 +3355,7 @@ func TestEntryOrdering(t *testing.T) {
 				},
 			},
 			assignment: flavorassigner.Assignment{
-				Borrowing: true,
+				Borrowing: 1,
 			},
 		},
 		{
@@ -3315,6 +3378,19 @@ func TestEntryOrdering(t *testing.T) {
 				},
 			},
 		},
+		{
+			Info: workload.Info{
+				Obj: &kueue.Workload{ObjectMeta: metav1.ObjectMeta{
+					Name:              "high_pri_borrowing_more",
+					CreationTimestamp: metav1.NewTime(now.Add(3 * time.Second)),
+				}, Spec: kueue.WorkloadSpec{
+					Priority: ptr.To[int32](1),
+				}},
+			},
+			assignment: flavorassigner.Assignment{
+				Borrowing: 2,
+			},
+		},
 	}
 	inputForOrderingPreemptedWorkloads := []entry{
 		{
@@ -3407,28 +3483,28 @@ func TestEntryOrdering(t *testing.T) {
 			input:            input,
 			prioritySorting:  true,
 			workloadOrdering: workload.Ordering{PodsReadyRequeuingTimestamp: config.EvictionTimestamp},
-			wantOrder:        []string{"new_high_pri", "old", "recently_evicted", "new", "high_pri_borrowing", "old_borrowing", "evicted_borrowing", "new_borrowing"},
+			wantOrder:        []string{"new_high_pri", "old", "recently_evicted", "new", "high_pri_borrowing", "old_borrowing", "evicted_borrowing", "new_borrowing", "high_pri_borrowing_more"},
 		},
 		{
 			name:             "Priority sorting is enabled (default) using pods-ready Creation timestamp",
 			input:            input,
 			prioritySorting:  true,
 			workloadOrdering: workload.Ordering{PodsReadyRequeuingTimestamp: config.CreationTimestamp},
-			wantOrder:        []string{"new_high_pri", "recently_evicted", "old", "new", "high_pri_borrowing", "old_borrowing", "evicted_borrowing", "new_borrowing"},
+			wantOrder:        []string{"new_high_pri", "recently_evicted", "old", "new", "high_pri_borrowing", "old_borrowing", "evicted_borrowing", "new_borrowing", "high_pri_borrowing_more"},
 		},
 		{
 			name:             "Priority sorting is disabled using pods-ready Eviction timestamp",
 			input:            input,
 			prioritySorting:  false,
 			workloadOrdering: workload.Ordering{PodsReadyRequeuingTimestamp: config.EvictionTimestamp},
-			wantOrder:        []string{"old", "recently_evicted", "new", "new_high_pri", "old_borrowing", "evicted_borrowing", "high_pri_borrowing", "new_borrowing"},
+			wantOrder:        []string{"old", "recently_evicted", "new", "new_high_pri", "old_borrowing", "evicted_borrowing", "high_pri_borrowing", "new_borrowing", "high_pri_borrowing_more"},
 		},
 		{
 			name:             "Priority sorting is disabled using pods-ready Creation timestamp",
 			input:            input,
 			prioritySorting:  false,
 			workloadOrdering: workload.Ordering{PodsReadyRequeuingTimestamp: config.CreationTimestamp},
-			wantOrder:        []string{"recently_evicted", "old", "new", "new_high_pri", "old_borrowing", "evicted_borrowing", "high_pri_borrowing", "new_borrowing"},
+			wantOrder:        []string{"recently_evicted", "old", "new", "new_high_pri", "old_borrowing", "evicted_borrowing", "high_pri_borrowing", "new_borrowing", "high_pri_borrowing_more"},
 		},
 		{
 			name:            "Some workloads are preempted; Priority sorting is disabled",
@@ -4149,7 +4225,7 @@ func TestResourcesToReserve(t *testing.T) {
 	cases := []struct {
 		name            string
 		assignmentMode  flavorassigner.FlavorAssignmentMode
-		borrowing       bool
+		borrowing       int
 		assignmentUsage resources.FlavorResourceQuantities
 		cqUsage         resources.FlavorResourceQuantities
 		wantReserved    resources.FlavorResourceQuantities
@@ -4229,7 +4305,7 @@ func TestResourcesToReserve(t *testing.T) {
 		{
 			name:           "Reserved memory cut by nominal+borrowing quota, assignment preempts and borrows",
 			assignmentMode: flavorassigner.Preempt,
-			borrowing:      true,
+			borrowing:      1,
 			assignmentUsage: resources.FlavorResourceQuantities{
 				{Flavor: kueue.ResourceFlavorReference("spot"), Resource: corev1.ResourceMemory}: 50,
 				{Flavor: kueue.ResourceFlavorReference("model-b"), Resource: "gpu"}:              2,
@@ -4248,7 +4324,7 @@ func TestResourcesToReserve(t *testing.T) {
 		{
 			name:           "Reserved memory equal assignment usage, CQ borrowing limit is nil",
 			assignmentMode: flavorassigner.Preempt,
-			borrowing:      true,
+			borrowing:      1,
 			assignmentUsage: resources.FlavorResourceQuantities{
 				{Flavor: kueue.ResourceFlavorReference("on-demand"), Resource: corev1.ResourceMemory}: 50,
 				{Flavor: kueue.ResourceFlavorReference("model-b"), Resource: "gpu"}:                   2,
diff --git a/test/integration/singlecluster/scheduler/preemption_test.go b/test/integration/singlecluster/scheduler/preemption_test.go
index 1cbe0ba5a..237c8c344 100644
--- a/test/integration/singlecluster/scheduler/preemption_test.go
+++ b/test/integration/singlecluster/scheduler/preemption_test.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -910,4 +911,81 @@ var _ = ginkgo.Describe("Preemption", func() {
 			})
 		})
 	})
+	ginkgo.Context("In a multi-level cohort", func() {
+		var rootCohort, guaranteedCohort *kueuealpha.Cohort
+		var bestEffortCQ, guaranteedCQ *kueue.ClusterQueue
+		var defaultFlavor *kueue.ResourceFlavor
+
+		ginkgo.BeforeEach(func() {
+			defaultFlavor = testing.MakeResourceFlavor("default").Obj()
+			gomega.Expect(k8sClient.Create(ctx, defaultFlavor)).To(gomega.Succeed())
+
+			rootCohort = testing.MakeCohort("root").Obj()
+			gomega.Expect(k8sClient.Create(ctx, rootCohort)).To(gomega.Succeed())
+			guaranteedCohort = testing.MakeCohort("guaranteed").Parent("root").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "3").Obj(),
+				).Obj()
+			gomega.Expect(k8sClient.Create(ctx, guaranteedCohort)).To(gomega.Succeed())
+			bestEffortCQ = testing.MakeClusterQueue("best-effort").
+				Cohort("root").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "0").Obj(),
+				).
+				Preemption(kueue.ClusterQueuePreemption{
+					WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+					ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+				}).
+				Obj()
+			gomega.Expect(k8sClient.Create(ctx, bestEffortCQ)).To(gomega.Succeed())
+			guaranteedCQ = testing.MakeClusterQueue("guaranteed").
+				Cohort("guaranteed").ResourceGroup(
+				*testing.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "0").Obj(),
+			).
+				Preemption(kueue.ClusterQueuePreemption{
+					WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
+					ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+				}).
+				Obj()
+			gomega.Expect(k8sClient.Create(ctx, guaranteedCQ)).To(gomega.Succeed())
+		})
+		ginkgo.It("workloads in guaranteed cq should have preferential access to the resources", func() {
+			var bestEffortLQ *kueue.LocalQueue
+			var guaranteedLQ *kueue.LocalQueue
+			var bestEffortWl, guaranteedWl *kueue.Workload
+
+			ginkgo.By("Create local queues", func() {
+				bestEffortLQ = testing.MakeLocalQueue("best-effort-lq", ns.Name).ClusterQueue(bestEffortCQ.Name).Obj()
+				gomega.Expect(k8sClient.Create(ctx, bestEffortLQ)).Should(gomega.Succeed())
+				guaranteedLQ = testing.MakeLocalQueue("guaranteed-lq", ns.Name).ClusterQueue(guaranteedCQ.Name).Obj()
+				gomega.Expect(k8sClient.Create(ctx, guaranteedLQ)).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Create a workload in bestEffortCQ and await for admission", func() {
+				bestEffortWl = testing.MakeWorkload("best-effort-wl", ns.Name).
+					Queue(bestEffortLQ.Name).
+					Priority(highPriority).
+					Request(corev1.ResourceCPU, "3").
+					Obj()
+				gomega.Expect(k8sClient.Create(ctx, bestEffortWl)).To(gomega.Succeed())
+				util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, bestEffortWl)
+			})
+
+			ginkgo.By("Create a workload in guaranteedCQ", func() {
+				guaranteedWl = testing.MakeWorkload("guaranteed-wl", ns.Name).
+					Queue(guaranteedLQ.Name).
+					Priority(midPriority).
+					Request(corev1.ResourceCPU, "3").
+					Obj()
+				gomega.Expect(k8sClient.Create(ctx, guaranteedWl)).To(gomega.Succeed())
+			})
+
+			ginkgo.By("Await for preemption of the workload in bestEffortCQ and admission of guaranteedWL", func() {
+				util.ExpectWorkloadsToBePreempted(ctx, k8sClient, bestEffortWl)
+				util.FinishEvictionForWorkloads(ctx, k8sClient, bestEffortWl)
+				util.ExpectWorkloadsToBePending(ctx, k8sClient, bestEffortWl)
+				util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, guaranteedWl)
+			})
+		})
+	})
 })
diff --git a/test/integration/singlecluster/scheduler/workload_controller_test.go b/test/integration/singlecluster/scheduler/workload_controller_test.go
index 58555997e..489ae502e 100644
--- a/test/integration/singlecluster/scheduler/workload_controller_test.go
+++ b/test/integration/singlecluster/scheduler/workload_controller_test.go
@@ -140,7 +140,7 @@ var _ = ginkgo.Describe("Workload controller with scheduler", func() {
 					g.Expect(workload.HasQuotaReservation(&updatedWl)).Should(gomega.BeTrue(), "should have quota reservation")
 
 					checks := slices.Map(updatedWl.Status.AdmissionChecks, func(c *kueue.AdmissionCheckState) kueue.AdmissionCheckReference { return c.Name })
-					g.Expect(checks).Should(gomega.ConsistOf("check1", "check2"))
+					g.Expect(checks).Should(gomega.ConsistOf(kueue.AdmissionCheckReference("check1"), kueue.AdmissionCheckReference("check2")))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 				gomega.Expect(workload.IsAdmitted(&updatedWl)).To(gomega.BeFalse())
 			})
@@ -157,7 +157,7 @@ var _ = ginkgo.Describe("Workload controller with scheduler", func() {
 					g.Expect(k8sClient.Update(ctx, &createdQueue)).To(gomega.Succeed())
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
 					checks := slices.Map(updatedWl.Status.AdmissionChecks, func(c *kueue.AdmissionCheckState) kueue.AdmissionCheckReference { return c.Name })
-					g.Expect(checks).Should(gomega.ConsistOf("check1", "check3"))
+					g.Expect(checks).Should(gomega.ConsistOf(kueue.AdmissionCheckReference("check1"), kueue.AdmissionCheckReference("check3")))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})