Repository: kubernetes-sigs/kueue

Issue:
Extend waitForPodsReady to evict a workload if a replacement pod cannot schedule 
**What would you like to be added**:

Extend the waitForPodsReady mechanism to allow workload eviction if there is a replacement pod which cannot be scheduled within a specified timeout.

**Why is this needed**:

Currently the mechanism only evicts workloads if they cannot start, but a pod can fail while the  workload is running.

**Completion requirements**:

This enhancement requires the following artifacts:

- [x] Design doc
- [x] API change
- [ ] Docs update

The artifacts should be linked in subsequent comments.


Hints from issue comments:
@mimowo: /assign @PBundyra @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/@mimowo: /remove-lifecycle stale@yaroslava-serdiuk: cc: @yaroslava-serdiuk @tenzen-y: /reopen

@PBundyra could you extend documentation in https://kueue.sigs.k8s.io/docs/tasks/manage/setup_wait_for_pods_ready/?
@k8s-ci-robot: @tenzen-y: Reopened this issue.

<details>

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

>/reopen
>
>@PBundyra could you extend documentation in https://kueue.sigs.k8s.io/docs/tasks/manage/setup_wait_for_pods_ready/?
>


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>@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/@PBundyra: The field has been documented a while ago with this PR https://github.com/kubernetes-sigs/kueue/pull/4964
I believe we can close this issue as it was addressed
/close
@k8s-ci-robot: @PBundyra: Closing this issue.

<details>

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

>The field has been documented a while ago with this PR https://github.com/kubernetes-sigs/kueue/pull/4964
>I believe we can close this issue as it was addressed
>/close
>


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/tas_flavor.go b/pkg/cache/tas_flavor.go
index 8b65bcd98..1a53a6af7 100644
--- a/pkg/cache/tas_flavor.go
+++ b/pkg/cache/tas_flavor.go
@@ -163,8 +163,10 @@ func (c *TASFlavorCache) updateUsage(topologyRequests []workload.TopologyDomainR
 		}
 		if op == subtract {
 			c.usage[domainID].Sub(tr.Requests)
+			c.usage[domainID].Sub(resources.Requests{corev1.ResourcePods: int64(tr.Count)})
 		} else {
 			c.usage[domainID].Add(tr.Requests)
+			c.usage[domainID].Add(resources.Requests{corev1.ResourcePods: int64(tr.Count)})
 		}
 	}
 }
diff --git a/pkg/cache/tas_flavor_snapshot.go b/pkg/cache/tas_flavor_snapshot.go
index b6611219c..909c888b1 100644
--- a/pkg/cache/tas_flavor_snapshot.go
+++ b/pkg/cache/tas_flavor_snapshot.go
@@ -209,6 +209,7 @@ func (s *TASFlavorSnapshot) addNonTASUsage(domainID utiltas.TopologyDomainID, us
 	// least one TAS pod, and so the addCapacity function to initialize
 	// freeCapacity is already called.
 	s.leaves[domainID].freeCapacity.Sub(usage)
+	s.leaves[domainID].freeCapacity.Sub(resources.Requests{corev1.ResourcePods: 1})
 }
 
 func (s *TASFlavorSnapshot) addTASUsage(domainID utiltas.TopologyDomainID, usage resources.Requests) {
@@ -299,6 +300,7 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	assumedUsage map[utiltas.TopologyDomainID]resources.Requests) (*kueue.TopologyAssignment, string) {
 	topologyRequest := tasPodSetRequests.PodSet.TopologyRequest
 	requests := tasPodSetRequests.SinglePodRequests
+	requests.Add(resources.Requests{corev1.ResourcePods: 1})
 	podSetTolerations := tasPodSetRequests.PodSet.Template.Spec.Tolerations
 	count := tasPodSetRequests.Count
 	required := topologyRequest.Required != nil
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 83c9fd5bc..3906bc831 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -830,11 +830,7 @@ func (r *WorkloadReconciler) admittedNotReadyWorkload(wl *kueue.Workload) (bool,
 	if podsReadyCond != nil && podsReadyCond.Status == metav1.ConditionFalse && podsReadyCond.LastTransitionTime.After(admittedCond.LastTransitionTime.Time) {
 		elapsedTime = r.clock.Since(podsReadyCond.LastTransitionTime.Time)
 	}
-	waitFor := r.waitForPodsReady.timeout - elapsedTime
-	if waitFor < 0 {
-		waitFor = 0
-	}
-	return true, waitFor
+	return true, max(r.waitForPodsReady.timeout-elapsedTime, 0)
 }
 
 type resourceUpdatesHandler struct {
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index 142cd10b3..7519d0cba 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -291,39 +291,39 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	isStandaloneJob := true
+	isTopLevelJob := true
 	objectOwner := metav1.GetControllerOf(object)
 	if objectOwner != nil && IsOwnerManagedByKueue(objectOwner) {
-		isStandaloneJob = false
+		isTopLevelJob = false
 	}
 
 	// when manageJobsWithoutQueueName is disabled we only reconcile jobs that either
-	// have a queue-name label or have a kueue-managed parent that has a queue-name label.
+	// have a queue-name label or have a kueue-managed ancestor that has a queue-name label.
 	if !r.manageJobsWithoutQueueName && QueueName(job) == "" {
-		if isStandaloneJob {
+		if isTopLevelJob {
 			log.V(3).Info("queue-name label is not set, ignoring the job", "queueName", QueueName(job))
 			return ctrl.Result{}, nil
 		}
-		isParentJobManaged, err := r.IsParentJobManaged(ctx, job.Object(), req.Namespace)
+		isAncestorJobManaged, err := r.IsAncestorJobManaged(ctx, job.Object(), req.Namespace)
 		if err != nil {
-			log.Error(err, "couldn't check whether the parent job is managed by kueue")
+			log.Error(err, "couldn't check whether an ancestor job is managed by kueue")
 			return ctrl.Result{}, err
 		}
-		if !isParentJobManaged {
-			log.V(3).Info("parent job is managed by kueue but doesn't have a queue-name label, ignoring the job",
+		if !isAncestorJobManaged {
+			log.V(3).Info("No kueue-managed ancestors have a queue-name label, ignoring the job",
 				"parentJob", objectOwner.Name)
 			return ctrl.Result{}, nil
 		}
 	}
 
-	// if this is a non-standalone job, suspend the job if its parent workload is not found or not admitted.
-	if !isStandaloneJob {
+	// if this is a non-toplevel job, suspend the job if its ancestor's workload is not found or not admitted.
+	if !isTopLevelJob {
 		_, _, finished := job.Finished()
 		if !finished && !job.IsSuspended() {
-			if parentWorkload, err := r.getParentWorkload(ctx, object); err != nil {
-				log.Error(err, "couldn't get the parent job workload")
+			if ancestorWorkload, err := r.getAncestorWorkload(ctx, object, req.Namespace); err != nil {
+				log.Error(err, "couldn't get an ancestor job workload")
 				return ctrl.Result{}, err
-			} else if parentWorkload == nil || !workload.IsAdmitted(parentWorkload) {
+			} else if ancestorWorkload == nil || !workload.IsAdmitted(ancestorWorkload) {
 				if err := clientutil.Patch(ctx, r.client, object, true, func() (bool, error) {
 					job.Suspend()
 					return true, nil
@@ -569,34 +569,60 @@ func (r *JobReconciler) recordAdmissionCheckUpdate(wl *kueue.Workload, job Gener
 	}
 }
 
-// IsParentJobManaged checks whether the parent job is managed by kueue.
-func (r *JobReconciler) IsParentJobManaged(ctx context.Context, jobObj client.Object, namespace string) (bool, error) {
-	owner := metav1.GetControllerOf(jobObj)
-	if owner == nil {
-		return false, nil
-	}
-	parentJob := GetEmptyOwnerObject(owner)
-	if parentJob == nil {
-		return false, fmt.Errorf("workload owner %v: %w", owner, ErrUnknownWorkloadOwner)
-	}
-	if err := r.client.Get(ctx, client.ObjectKey{Name: owner.Name, Namespace: namespace}, parentJob); err != nil {
-		return false, errors.Join(ErrWorkloadOwnerNotFound, err)
+// IsAncestorJobManaged checks whether an ancestor job is managed by kueue.
+func (r *JobReconciler) IsAncestorJobManaged(ctx context.Context, jobObj client.Object, namespace string) (bool, error) {
+	ancestor, err := r.getAncestorJobManagedByKueue(ctx, jobObj, namespace)
+	if err != nil {
+		return false, err
 	}
-	return QueueNameForObject(parentJob) != "", nil
+	return ancestor != nil, nil
 }
 
-func (r *JobReconciler) getParentWorkload(ctx context.Context, object client.Object) (*kueue.Workload, error) {
-	owner := metav1.GetControllerOf(object)
-	if owner == nil {
+// getAncestorWorkload returns the Workload object of the Kueue-managed ancestor job.
+func (r *JobReconciler) getAncestorWorkload(ctx context.Context, jobObj client.Object, namespace string) (*kueue.Workload, error) {
+	ancestor, err := r.getAncestorJobManagedByKueue(ctx, jobObj, namespace)
+	if err != nil {
+		return nil, err
+	}
+	if ancestor == nil {
 		return nil, nil
 	}
 	wlList := kueue.WorkloadList{}
-	if err := r.client.List(ctx, &wlList, client.InNamespace(object.GetNamespace()), client.MatchingFields{indexer.OwnerReferenceUID: string(owner.UID)}); err != nil || len(wlList.Items) == 0 {
-		return nil, client.IgnoreNotFound(err)
+	if err := r.client.List(ctx, &wlList, client.InNamespace(ancestor.GetNamespace()), client.MatchingFields{indexer.OwnerReferenceUID: string(ancestor.GetUID())}); client.IgnoreNotFound(err) != nil {
+		return nil, err
+	}
+	if len(wlList.Items) > 0 {
+		// In theory the job can own multiple Workloads, we cannot do too much about it, maybe log it.
+		return &wlList.Items[0], nil
 	}
+	return nil, nil
+}
 
-	// In theory the parent can own multiple Workloads, we cannot do too much about it, maybe log it.
-	return &wlList.Items[0], nil
+// getAncestorJobManagedByKueue traverses controllerRefs to find an ancestor job that is manged by Kueue (ie, it has a queue-name label).
+func (r *JobReconciler) getAncestorJobManagedByKueue(ctx context.Context, jobObj client.Object, namespace string) (client.Object, error) {
+	seen := make(sets.Set[types.UID])
+	for {
+		if seen.Has(jobObj.GetUID()) {
+			return nil, nil
+		}
+		seen.Insert(jobObj.GetUID())
+
+		owner := metav1.GetControllerOf(jobObj)
+		if owner == nil || !IsOwnerManagedByKueue(owner) {
+			return nil, nil
+		}
+		parentJob := GetEmptyOwnerObject(owner)
+		if parentJob == nil {
+			return nil, fmt.Errorf("workload owner %v: %w", owner, ErrUnknownWorkloadOwner)
+		}
+		if err := r.client.Get(ctx, client.ObjectKey{Name: owner.Name, Namespace: namespace}, parentJob); err != nil {
+			return nil, errors.Join(ErrWorkloadOwnerNotFound, err)
+		}
+		if QueueNameForObject(parentJob) != "" {
+			return parentJob, nil
+		}
+		jobObj = parentJob
+	}
 }
 
 // ensureOneWorkload will query for the single matched workload corresponding to job and return it.
diff --git a/pkg/controller/jobs/job/job_controller.go b/pkg/controller/jobs/job/job_controller.go
index 6ef41e8ab..c8b833408 100644
--- a/pkg/controller/jobs/job/job_controller.go
+++ b/pkg/controller/jobs/job/job_controller.go
@@ -314,7 +314,11 @@ func (j *Job) Finished() (message string, success, finished bool) {
 
 func (j *Job) PodsReady() bool {
 	ready := ptr.Deref(j.Status.Ready, 0)
-	return j.Status.Succeeded+ready >= j.podsCount()
+	uncountedTerminatedSucceeded := 0
+	if j.Status.UncountedTerminatedPods != nil {
+		uncountedTerminatedSucceeded = len(j.Status.UncountedTerminatedPods.Succeeded)
+	}
+	return j.Status.Succeeded+ready+int32(uncountedTerminatedSucceeded) >= j.podsCount()
 }
 
 func (j *Job) CanDefaultManagedBy() bool {
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 54ca84f52..3f8e62261 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -182,6 +182,8 @@ type TopologyRequest struct {
 type TopologyDomainRequests struct {
 	Values   []string
 	Requests resources.Requests
+	// Count indicates how many pods are requested in this TopologyDomain.
+	Count int32
 }
 
 func (psr *PodSetResources) ScaledTo(newCount int32) *PodSetResources {
@@ -414,6 +416,7 @@ func totalRequestsFromAdmission(wl *kueue.Workload) []PodSetResources {
 				setRes.TopologyRequest.DomainRequests = append(setRes.TopologyRequest.DomainRequests, TopologyDomainRequests{
 					Values:   domain.Values,
 					Requests: domainRequests,
+					Count:    domain.Count,
 				})
 			}
 		}



Test patch:
diff --git a/pkg/cache/tas_cache_test.go b/pkg/cache/tas_cache_test.go
index cd8c2c46c..1d7e31324 100644
--- a/pkg/cache/tas_cache_test.go
+++ b/pkg/cache/tas_cache_test.go
@@ -55,6 +55,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -65,6 +66,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -75,6 +77,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -85,6 +88,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -95,6 +99,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -105,6 +110,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("2"),
 				corev1.ResourceMemory: resource.MustParse("4Gi"),
+				corev1.ResourcePods:   resource.MustParse("40"),
 			}).
 			Ready().
 			Obj(),
@@ -135,6 +141,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -145,6 +152,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -155,6 +163,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -165,6 +174,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -175,6 +185,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -185,6 +196,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -195,6 +207,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -205,6 +218,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -241,7 +255,8 @@ func TestFindTopologyAssignment(t *testing.T) {
 					Label(tasRackLabel, "r1").
 					Label(corev1.LabelHostname, "x1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("2"),
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -250,7 +265,8 @@ func TestFindTopologyAssignment(t *testing.T) {
 					Label(tasRackLabel, "r2").
 					Label(corev1.LabelHostname, "x2").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("2"),
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("20"),
 					}).
 					Ready().
 					Obj(),
@@ -259,7 +275,8 @@ func TestFindTopologyAssignment(t *testing.T) {
 					Label(tasRackLabel, "r3").
 					Label(corev1.LabelHostname, "x3").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -268,7 +285,8 @@ func TestFindTopologyAssignment(t *testing.T) {
 					Label(tasRackLabel, "r3").
 					Label(corev1.LabelHostname, "x4").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -277,7 +295,8 @@ func TestFindTopologyAssignment(t *testing.T) {
 					Label(tasRackLabel, "r3").
 					Label(corev1.LabelHostname, "x5").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -286,7 +305,8 @@ func TestFindTopologyAssignment(t *testing.T) {
 					Label(tasRackLabel, "r3").
 					Label(corev1.LabelHostname, "x6").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -537,21 +557,24 @@ func TestFindTopologyAssignment(t *testing.T) {
 				*testingnode.MakeNode("b1").
 					Label(tasBlockLabel, "b1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("2"),
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("20"),
 					}).
 					Ready().
 					Obj(),
 				*testingnode.MakeNode("b2").
 					Label(tasBlockLabel, "b2").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
 				*testingnode.MakeNode("b3").
 					Label(tasBlockLabel, "b3").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("4"),
+						corev1.ResourceCPU:  resource.MustParse("4"),
+						corev1.ResourcePods: resource.MustParse("40"),
 					}).
 					Ready().
 					Obj(),
@@ -919,6 +942,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Obj(),
 			},
@@ -944,6 +968,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -981,6 +1006,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1004,6 +1030,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1041,6 +1068,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1084,6 +1112,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1113,6 +1142,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1143,6 +1173,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1151,6 +1182,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1189,6 +1221,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					NotReady().
 					StatusConditions(corev1.NodeCondition{
@@ -1219,6 +1252,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Unschedulable().
@@ -1246,6 +1280,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Taints(corev1.Taint{
@@ -1282,6 +1317,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 					StatusAllocatable(corev1.ResourceList{
 						corev1.ResourceCPU:    resource.MustParse("1"),
 						corev1.ResourceMemory: resource.MustParse("1Gi"),
+						corev1.ResourcePods:   resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -1317,6 +1353,39 @@ func TestFindTopologyAssignment(t *testing.T) {
 			},
 			enableTASLeastAllocated: true,
 		},
+		"no assignment as node does not have enough allocatable pods (.status.allocatable['pods']); LeastAllocated": {
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label("zone", "zone-a").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1000m"),
+						corev1.ResourcePods: resource.MustParse("1"),
+					}).
+					Ready().
+					Obj(),
+			},
+			pods: []corev1.Pod{
+				*testingpod.MakePod("test-running", "test-ns").
+					NodeName("b1-r1-x1").
+					StatusPhase(corev1.PodRunning).
+					Request(corev1.ResourceCPU, "300m").
+					Obj(),
+			},
+			request: kueue.PodSetTopologyRequest{
+				Required: ptr.To(corev1.LabelHostname),
+			},
+			nodeLabels: map[string]string{
+				"zone": "zone-a",
+			},
+			levels: defaultOneLevel,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 300,
+			},
+			count:                   1,
+			wantReason:              `topology "default" doesn't allow to fit any of 1 pod(s)`,
+			enableTASLeastAllocated: true,
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
diff --git a/pkg/controller/jobframework/reconciler_test.go b/pkg/controller/jobframework/reconciler_test.go
index 6dd8eb6d0..f51fa2e7c 100644
--- a/pkg/controller/jobframework/reconciler_test.go
+++ b/pkg/controller/jobframework/reconciler_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	kfmpi "github.com/kubeflow/mpi-operator/pkg/apis/kubeflow/v2beta1"
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -34,6 +35,7 @@ import (
 	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
 	"sigs.k8s.io/kueue/pkg/util/kubeversion"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	testingaw "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingmpijob "sigs.k8s.io/kueue/pkg/util/testingjobs/mpijob"
 
@@ -42,24 +44,26 @@ import (
 	. "sigs.k8s.io/kueue/pkg/controller/jobframework"
 )
 
-func TestIsParentJobManaged(t *testing.T) {
+func TestIsAncestorJobManaged(t *testing.T) {
+	grandparentJobName := "test-job-grandparent"
 	parentJobName := "test-job-parent"
 	childJobName := "test-job-child"
 	jobNamespace := "default"
 	cases := map[string]struct {
-		parentJob   client.Object
-		job         client.Object
-		wantManaged bool
-		wantErr     error
+		grandparentJob client.Object
+		parentJob      client.Object
+		job            client.Object
+		wantManaged    bool
+		wantErr        error
 	}{
-		"child job has ownerReference with unknown workload owner": {
+		"child job has ownerReference with unmanaged workload owner": {
 			parentJob: testingjob.MakeJob(parentJobName, jobNamespace).
 				UID(parentJobName).
 				Obj(),
 			job: testingjob.MakeJob(childJobName, jobNamespace).
 				OwnerReference(parentJobName, batchv1.SchemeGroupVersion.WithKind("CronJob")).
 				Obj(),
-			wantErr: ErrUnknownWorkloadOwner,
+			wantManaged: false,
 		},
 		"child job has ownerReference with known non-existing workload owner": {
 			job: testingjob.MakeJob(childJobName, jobNamespace).
@@ -85,17 +89,64 @@ func TestIsParentJobManaged(t *testing.T) {
 				OwnerReference(parentJobName, kfmpi.SchemeGroupVersionKind).
 				Obj(),
 		},
+		"child job has managed parent and grandparent and grandparent has a queue-name label": {
+			grandparentJob: testingaw.MakeAppWrapper(grandparentJobName, jobNamespace).
+				UID(grandparentJobName).
+				Queue("test-q").
+				Obj(),
+			parentJob: testingmpijob.MakeMPIJob(parentJobName, jobNamespace).
+				UID(parentJobName).
+				OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind("AppWrapper")).
+				Obj(),
+			job: testingjob.MakeJob(childJobName, jobNamespace).
+				OwnerReference(parentJobName, kfmpi.SchemeGroupVersionKind).
+				Obj(),
+			wantManaged: true,
+		},
+		"child job has managed parent and grandparent and grandparent doesn't have a queue-name label": {
+			grandparentJob: testingaw.MakeAppWrapper(grandparentJobName, jobNamespace).
+				UID(grandparentJobName).
+				Obj(),
+			parentJob: testingmpijob.MakeMPIJob(parentJobName, jobNamespace).
+				UID(parentJobName).
+				OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind("AppWrapper")).
+				Obj(),
+			job: testingjob.MakeJob(childJobName, jobNamespace).
+				OwnerReference(parentJobName, kfmpi.SchemeGroupVersionKind).
+				Obj(),
+			wantManaged: false,
+		},
+		"cyclic ownership links are properly handled": {
+			grandparentJob: testingaw.MakeAppWrapper(grandparentJobName, jobNamespace).
+				UID(grandparentJobName).
+				OwnerReference(childJobName, batchv1.SchemeGroupVersion.WithKind("Job")).
+				Obj(),
+			parentJob: testingmpijob.MakeMPIJob(parentJobName, jobNamespace).
+				UID(parentJobName).
+				OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind("AppWrapper")).
+				Obj(),
+			job: testingjob.MakeJob(childJobName, jobNamespace).
+				OwnerReference(parentJobName, kfmpi.SchemeGroupVersionKind).
+				Obj(),
+			wantManaged: false,
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
-			t.Cleanup(EnableIntegrationsForTest(t, "kubeflow.org/mpijob"))
-			builder := utiltesting.NewClientBuilder(kfmpi.AddToScheme)
+			t.Cleanup(EnableIntegrationsForTest(t, "kubeflow.org/mpijob", "workload.codeflare.dev/appwrapper", "batch/job"))
+			builder := utiltesting.NewClientBuilder(kfmpi.AddToScheme, awv1beta2.AddToScheme)
+			if tc.grandparentJob != nil {
+				builder = builder.WithObjects(tc.grandparentJob)
+			}
 			if tc.parentJob != nil {
 				builder = builder.WithObjects(tc.parentJob)
 			}
+			if tc.job != nil {
+				builder = builder.WithObjects(tc.job)
+			}
 			cl := builder.Build()
 			r := NewReconciler(cl, nil)
-			got, gotErr := r.IsParentJobManaged(context.Background(), tc.job, jobNamespace)
+			got, gotErr := r.IsAncestorJobManaged(context.Background(), tc.job, jobNamespace)
 			if tc.wantManaged != got {
 				t.Errorf("Unexpected response from isParentManaged want: %v,got: %v", tc.wantManaged, got)
 			}
diff --git a/pkg/controller/jobs/job/job_controller_test.go b/pkg/controller/jobs/job/job_controller_test.go
index a6462b7c9..c8e3f967f 100644
--- a/pkg/controller/jobs/job/job_controller_test.go
+++ b/pkg/controller/jobs/job/job_controller_test.go
@@ -1896,6 +1896,7 @@ func TestReconciler(t *testing.T) {
 			otherJobs: []batchv1.Job{
 				*utiltestingjob.MakeJob("parent", "ns").
 					Queue("queue").
+					UID("parent-uid").
 					Obj(),
 			},
 			workloads: []kueue.Workload{
@@ -1903,7 +1904,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 10).SetMinimumCount(5).Request(corev1.ResourceCPU, "1").Obj()).
 					ReserveQuota(utiltesting.MakeAdmission("cq").AssignmentPodCount(10).Obj()).
 					Admitted(true).
-					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent").
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent-uid").
 					Obj(),
 			},
 			wantWorkloads: []kueue.Workload{
@@ -1911,7 +1912,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 10).SetMinimumCount(5).Request(corev1.ResourceCPU, "1").Obj()).
 					ReserveQuota(utiltesting.MakeAdmission("cq").AssignmentPodCount(10).Obj()).
 					Admitted(true).
-					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent").
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent-uid").
 					Obj(),
 			},
 		},
@@ -1932,20 +1933,21 @@ func TestReconciler(t *testing.T) {
 			otherJobs: []batchv1.Job{
 				*utiltestingjob.MakeJob("parent", "ns").
 					Queue("queue").
+					UID("parent-uid").
 					Obj(),
 			},
 			workloads: []kueue.Workload{
 				*utiltesting.MakeWorkload("parent-workload", "ns").
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 10).SetMinimumCount(5).Request(corev1.ResourceCPU, "1").Obj()).
-					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent").
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent-uid").
 					Obj(),
 			},
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload("parent-workload", "ns").
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 10).SetMinimumCount(5).Request(corev1.ResourceCPU, "1").Obj()).
-					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent").
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent-uid").
 					Obj(),
 			},
 			wantEvents: []utiltesting.EventRecord{
@@ -1972,6 +1974,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			otherJobs: []batchv1.Job{
 				*utiltestingjob.MakeJob("parent", "ns").
+					UID("parent-uid").
 					Queue("queue").
 					Obj(),
 			},
@@ -1980,7 +1983,7 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 10).SetMinimumCount(5).Request(corev1.ResourceCPU, "1").Obj()).
 					ReserveQuota(utiltesting.MakeAdmission("cq").AssignmentPodCount(10).Obj()).
-					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent").
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent-uid").
 					Admitted(true).
 					Obj(),
 			},
@@ -1989,7 +1992,7 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 10).SetMinimumCount(5).Request(corev1.ResourceCPU, "1").Obj()).
 					ReserveQuota(utiltesting.MakeAdmission("cq").AssignmentPodCount(10).Obj()).
-					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent").
+					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "parent", "parent-uid").
 					Admitted(true).
 					Obj(),
 			},
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 91df9f31a..79e6a2c59 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -3836,6 +3836,7 @@ func TestScheduleForTAS(t *testing.T) {
 			StatusAllocatable(corev1.ResourceList{
 				corev1.ResourceCPU:    resource.MustParse("1"),
 				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
 			}).
 			Ready().
 			Obj(),
@@ -4135,7 +4136,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label("tas-node", "true").
 					Label("cloud.com/custom-level", "x1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4462,7 +4464,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label(tasRackLabel, "r1").
 					Label(corev1.LabelHostname, "x1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("3"),
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4471,7 +4474,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label(tasRackLabel, "r1").
 					Label(corev1.LabelHostname, "y1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("3"),
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4518,7 +4522,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label(tasRackLabel, "r1").
 					Label(corev1.LabelHostname, "x1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("8"),
+						corev1.ResourceCPU:  resource.MustParse("8"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4527,7 +4532,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label(tasRackLabel, "r1").
 					Label(corev1.LabelHostname, "y1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("8"),
+						corev1.ResourceCPU:  resource.MustParse("8"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4612,7 +4618,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label(tasRackLabel, "r1").
 					Label(corev1.LabelHostname, "x1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("8"),
+						corev1.ResourceCPU:  resource.MustParse("8"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4621,7 +4628,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label(tasRackLabel, "r1").
 					Label(corev1.LabelHostname, "y1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("8"),
+						corev1.ResourceCPU:  resource.MustParse("8"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4708,7 +4716,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label("tas-node", "true").
 					Label(corev1.LabelHostname, "y1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Ready().
 					Obj(),
@@ -4793,7 +4802,8 @@ func TestScheduleForTAS(t *testing.T) {
 					Label("tas-node", "true").
 					Label(corev1.LabelHostname, "x1").
 					StatusAllocatable(corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("1"),
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
 					}).
 					Taints(corev1.Taint{
 						Key:    "example.com/gpu",
@@ -4914,6 +4924,73 @@ func TestScheduleForTAS(t *testing.T) {
 				},
 			},
 		},
+		"workload does not get scheduled as the node capacity (.status.allocatable['pods']) is already used by non-TAS and TAS workloads": {
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("x1").
+					Label("tas-node", "true").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1000m"),
+						corev1.ResourcePods: resource.MustParse("3"),
+					}).
+					Ready().
+					Obj(),
+			},
+			pods: []corev1.Pod{
+				*testingpod.MakePod("test-running", "test-ns").
+					NodeName("x1").
+					StatusPhase(corev1.PodRunning).
+					Request(corev1.ResourceCPU, "300m").
+					Obj(),
+			},
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueue},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo", "default").
+					Queue("tas-main").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "300m").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("bar-admitted", "default").
+					Queue("tas-main").
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "150m").
+							AssignmentPodCount(2).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 2,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 2).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "150m").
+						Obj()).
+					Obj(),
+			},
+			wantInadmissibleLeft: map[string][]string{
+				"tas-main": {"default/foo"},
+			},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "foo"},
+					EventType: "Warning",
+					Reason:    "Pending",
+					Message:   `couldn't assign flavors to pod set one: topology "tas-single-level" doesn't allow to fit any of 1 pod(s)`,
+				},
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
diff --git a/pkg/util/testingjobs/appwrapper/wrappers.go b/pkg/util/testingjobs/appwrapper/wrappers.go
index 5adda05f0..65e26e098 100644
--- a/pkg/util/testingjobs/appwrapper/wrappers.go
+++ b/pkg/util/testingjobs/appwrapper/wrappers.go
@@ -24,6 +24,9 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
 
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 )
@@ -86,6 +89,24 @@ func (aw *AppWrapperWrapper) Name(n string) *AppWrapperWrapper {
 	return aw
 }
 
+// UID updates the uid of the job.
+func (aw *AppWrapperWrapper) UID(uid string) *AppWrapperWrapper {
+	aw.ObjectMeta.UID = types.UID(uid)
+	return aw
+}
+
+// OwnerReference adds a ownerReference to the default container.
+func (j *AppWrapperWrapper) OwnerReference(ownerName string, ownerGVK schema.GroupVersionKind) *AppWrapperWrapper {
+	j.ObjectMeta.OwnerReferences = append(j.ObjectMeta.OwnerReferences, metav1.OwnerReference{
+		APIVersion: ownerGVK.GroupVersion().String(),
+		Kind:       ownerGVK.Kind,
+		Name:       ownerName,
+		UID:        types.UID(ownerName),
+		Controller: ptr.To(true),
+	})
+	return j
+}
+
 // Component adds a component to the AppWrapper
 func (aw *AppWrapperWrapper) Component(comp runtime.Object) *AppWrapperWrapper {
 	data, err := json.Marshal(comp)
diff --git a/pkg/util/testingjobs/mpijob/wrappers_mpijob.go b/pkg/util/testingjobs/mpijob/wrappers_mpijob.go
index 47291b345..7f9a6166d 100644
--- a/pkg/util/testingjobs/mpijob/wrappers_mpijob.go
+++ b/pkg/util/testingjobs/mpijob/wrappers_mpijob.go
@@ -21,6 +21,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/utils/ptr"
 
@@ -195,6 +196,18 @@ func (j *MPIJobWrapper) UID(uid string) *MPIJobWrapper {
 	return j
 }
 
+// OwnerReference adds a ownerReference to the default container.
+func (j *MPIJobWrapper) OwnerReference(ownerName string, ownerGVK schema.GroupVersionKind) *MPIJobWrapper {
+	j.ObjectMeta.OwnerReferences = append(j.ObjectMeta.OwnerReferences, metav1.OwnerReference{
+		APIVersion: ownerGVK.GroupVersion().String(),
+		Kind:       ownerGVK.Kind,
+		Name:       ownerName,
+		UID:        types.UID(ownerName),
+		Controller: ptr.To(true),
+	})
+	return j
+}
+
 // PodAnnotation sets annotation at the pod template level
 func (j *MPIJobWrapper) PodAnnotation(replicaType kfmpi.MPIReplicaType, k, v string) *MPIJobWrapper {
 	if j.Spec.MPIReplicaSpecs[replicaType].Template.Annotations == nil {
diff --git a/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go b/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
index f9856ed4a..07fcb9259 100644
--- a/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
@@ -880,6 +880,7 @@ var _ = ginkgo.Describe("AppWrapper controller when TopologyAwareScheduling enab
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
index e86c1ff17..232957906 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -1082,6 +1082,21 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 				Message: "All pods were ready or succeeded since the workload admission",
 			},
 		}),
+		ginkgo.Entry("One pod ready, one terminating succeeded", podsReadyTestSpec{
+			jobStatus: batchv1.JobStatus{
+				Active: 1,
+				Ready:  ptr.To[int32](1),
+				UncountedTerminatedPods: &batchv1.UncountedTerminatedPods{
+					Succeeded: []types.UID{"foo"},
+				},
+			},
+			wantCondition: &metav1.Condition{
+				Type:    kueue.WorkloadPodsReady,
+				Status:  metav1.ConditionTrue,
+				Reason:  "PodsReady",
+				Message: "All pods were ready or succeeded since the workload admission",
+			},
+		}),
 		ginkgo.Entry("One pod ready, one succeeded", podsReadyTestSpec{
 			jobStatus: batchv1.JobStatus{
 				Active:    1,
@@ -1095,6 +1110,33 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 				Message: "All pods were ready or succeeded since the workload admission",
 			},
 		}),
+		ginkgo.Entry("One pod succeeded, one terminating succeeded", podsReadyTestSpec{
+			jobStatus: batchv1.JobStatus{
+				Succeeded: 1,
+				UncountedTerminatedPods: &batchv1.UncountedTerminatedPods{
+					Succeeded: []types.UID{"foo"},
+				},
+			},
+			wantCondition: &metav1.Condition{
+				Type:    kueue.WorkloadPodsReady,
+				Status:  metav1.ConditionTrue,
+				Reason:  "PodsReady",
+				Message: "All pods were ready or succeeded since the workload admission",
+			},
+		}),
+		ginkgo.Entry("All pods terminating succeeded", podsReadyTestSpec{
+			jobStatus: batchv1.JobStatus{
+				UncountedTerminatedPods: &batchv1.UncountedTerminatedPods{
+					Succeeded: []types.UID{"foo", "bar"},
+				},
+			},
+			wantCondition: &metav1.Condition{
+				Type:    kueue.WorkloadPodsReady,
+				Status:  metav1.ConditionTrue,
+				Reason:  "PodsReady",
+				Message: "All pods were ready or succeeded since the workload admission",
+			},
+		}),
 		ginkgo.Entry("All pods are succeeded", podsReadyTestSpec{
 			jobStatus: batchv1.JobStatus{
 				Ready:     ptr.To[int32](0),
@@ -2288,6 +2330,7 @@ var _ = ginkgo.Describe("Job controller when TopologyAwareScheduling enabled", g
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go b/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go
index 6f0df5683..91b15c29d 100644
--- a/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go
@@ -1196,6 +1196,7 @@ var _ = ginkgo.Describe("JobSet controller when TopologyAwareScheduling enabled"
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go b/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go
index fd8d7dd9b..a99deadea 100644
--- a/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go
@@ -969,6 +969,7 @@ var _ = ginkgo.Describe("MPIJob controller when TopologyAwareScheduling enabled"
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go b/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go
index f250eb7bb..d0ead2c3a 100644
--- a/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go
@@ -348,6 +348,7 @@ var _ = ginkgo.Describe("PaddleJob controller when TopologyAwareScheduling enabl
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
index e8724906a..1e99d65b8 100644
--- a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
@@ -2204,6 +2204,7 @@ var _ = ginkgo.Describe("Pod controller when TopologyAwareScheduling enabled", g
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go b/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go
index 08894f238..adb68b06a 100644
--- a/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go
@@ -654,6 +654,7 @@ var _ = ginkgo.Describe("PyTorchJob controller when TopologyAwareScheduling enab
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go b/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go
index 982c784c3..1896d5d14 100644
--- a/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go
@@ -362,6 +362,7 @@ var _ = ginkgo.Describe("TFJob controller when TopologyAwareScheduling enabled",
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go b/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go
index 60aa1ee81..dde2cc4c2 100644
--- a/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go
@@ -345,6 +345,7 @@ var _ = ginkgo.Describe("XGBoostJob controller when TopologyAwareScheduling enab
 				StatusAllocatable(corev1.ResourceList{
 					corev1.ResourceCPU:    resource.MustParse("1"),
 					corev1.ResourceMemory: resource.MustParse("1Gi"),
+					corev1.ResourcePods:   resource.MustParse("10"),
 				}).
 				Ready().
 				Obj(),
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index 6a28c0020..0a6b0fd9e 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -333,6 +333,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -343,6 +344,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -353,6 +355,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -363,6 +366,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -734,6 +738,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -745,6 +750,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -756,6 +762,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -767,6 +774,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						StatusAllocatable(corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("1"),
 							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -920,6 +928,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 							StatusAllocatable(corev1.ResourceList{
 								corev1.ResourceCPU:    resource.MustParse("1"),
 								corev1.ResourceMemory: resource.MustParse("1Gi"),
+								corev1.ResourcePods:   resource.MustParse("10"),
 							}).
 							Ready().
 							Obj(),
@@ -993,6 +1002,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 							StatusAllocatable(corev1.ResourceList{
 								corev1.ResourceCPU:    resource.MustParse("1"),
 								corev1.ResourceMemory: resource.MustParse("1Gi"),
+								corev1.ResourcePods:   resource.MustParse("10"),
 							}).
 							Taints(corev1.Taint{
 								Key:    "maintenance",
@@ -1057,7 +1067,8 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						Label(corev1.LabelInstanceTypeStable, "cpu-node").
 						Label(testing.DefaultRackTopologyLevel, "cpu-rack").
 						StatusAllocatable(corev1.ResourceList{
-							corev1.ResourceCPU: resource.MustParse("5"),
+							corev1.ResourceCPU:  resource.MustParse("5"),
+							corev1.ResourcePods: resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),
@@ -1065,7 +1076,8 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 						Label(corev1.LabelInstanceTypeStable, "gpu-node").
 						Label(testing.DefaultRackTopologyLevel, "gpu-rack").
 						StatusAllocatable(corev1.ResourceList{
-							gpuResName: resource.MustParse("4"),
+							gpuResName:          resource.MustParse("4"),
+							corev1.ResourcePods: resource.MustParse("10"),
 						}).
 						Ready().
 						Obj(),