Repository: kubernetes-sigs/kueue

Issue:
Make Topology-Aware Scheduling compatible with cohorts and preemption
**What would you like to be added**:

Make TAS compatible with cohorts and preemption.

**Why is this needed**:

Currently, TAS does not work when cohorts or preemption are used. In that case the CQ is marked as inactive (see https://github.com/kubernetes-sigs/kueue/tree/main/keps/2724-topology-aware-scheduling#alpha). 

**Completion requirements**:

This enhancement requires the following artifacts:

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

The artifacts should be linked in subsequent comments.


Discussion:
@mimowo: cc @mwielgus @mwysokin @tenzen-y @mimowo: Listing cleanups to do identified during https://github.com/kubernetes-sigs/kueue/pull/4200:
- introduce PodSetReference to represent PodSet names for static code analysis
- add a unit test for `if len(cq.TASFlavors) == 0 {` as requested in [comment](https://github.com/kubernetes-sigs/kueue/pull/4200#discussion_r1951691235)
- add a unit test for `if cq.TASFlavors[*tasFlvr] == nil {` as requested in [comment](https://github.com/kubernetes-sigs/kueue/pull/4200#discussion_r1952616625)
- attempt to commonize the structures for `PodSetResources` and `TASPodSetRequests` as requested in [comment](https://github.com/kubernetes-sigs/kueue/pull/4200#discussion_r1950788801)@tenzen-y: Could you add https://github.com/kubernetes-sigs/kueue/pull/4200#discussion_r1952616625 to your list?
@mimowo: Sure, updated the list, but I think long-term it is better to manage in the refreshed [spreadsheet](https://docs.google.com/spreadsheets/d/1MXCjKZtAfqBTb61bJo46u7jIUqRIlu1NrYj1L8Xz-UU/edit?resourcekey=0-gr1ML2A1Axi8s6Lxr-Zhlw&gid=0#gid=0).@tenzen-y: > spreadsheet

Thanks, SGTM

Solution patch:
diff --git a/Makefile b/Makefile
index 1f45d8394..4ce1a78db 100644
--- a/Makefile
+++ b/Makefile
@@ -37,7 +37,7 @@ IMAGE_REGISTRY ?= $(STAGING_IMAGE_REGISTRY)/kueue
 IMAGE_NAME := kueue
 IMAGE_REPO ?= $(IMAGE_REGISTRY)/$(IMAGE_NAME)
 IMAGE_TAG ?= $(IMAGE_REPO):$(GIT_TAG)
-HELM_CHART_REPO := $(STAGING_IMAGE_REGISTRY)/charts
+HELM_CHART_REPO := $(STAGING_IMAGE_REGISTRY)/kueue/charts
 
 ifdef EXTRA_TAG
 IMAGE_EXTRA_TAG ?= $(IMAGE_REPO):$(EXTRA_TAG)
diff --git a/charts/kueue/Chart.yaml b/charts/kueue/Chart.yaml
index d52489db7..f2a1fc9ae 100644
--- a/charts/kueue/Chart.yaml
+++ b/charts/kueue/Chart.yaml
@@ -13,6 +13,9 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
+# NOTE: Do not modify manually. In Kueue, the version and appVersion are
+# overridden to GIT_TAG when building the artifacts, including the helm charts,
+# via Makefile.
 version: 0.1.0
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
diff --git a/hack/push-chart.sh b/hack/push-chart.sh
index dee97572a..e69fb2a7a 100755
--- a/hack/push-chart.sh
+++ b/hack/push-chart.sh
@@ -25,7 +25,7 @@ GIT_TAG=${GIT_TAG:-$(git describe --tags --dirty --always)}
 
 STAGING_IMAGE_REGISTRY=${STAGING_IMAGE_REGISTRY:-us-central1-docker.pkg.dev/k8s-staging-images}
 IMAGE_REGISTRY=${IMAGE_REGISTRY:-${STAGING_IMAGE_REGISTRY}/kueue}
-HELM_CHART_REPO=${HELM_CHART_REPO:-${STAGING_IMAGE_REGISTRY}/charts}
+HELM_CHART_REPO=${HELM_CHART_REPO:-${STAGING_IMAGE_REGISTRY}/kueue/charts}
 IMAGE_REPO=${IMAGE_REPO:-${IMAGE_REGISTRY}/kueue}
 
 HELM=${HELM:-./bin/helm}
diff --git a/pkg/cache/clusterqueue.go b/pkg/cache/clusterqueue.go
index fa88741fb..d73c26b79 100644
--- a/pkg/cache/clusterqueue.go
+++ b/pkg/cache/clusterqueue.go
@@ -282,10 +282,6 @@ func (c *clusterQueue) inactiveReason() (string, string) {
 				reasons = append(reasons, kueue.ClusterQueueActiveReasonNotSupportedWithTopologyAwareScheduling)
 				messages = append(messages, "TAS is not supported for cohorts")
 			}
-			if c.Preemption.WithinClusterQueue != kueue.PreemptionPolicyNever {
-				reasons = append(reasons, kueue.ClusterQueueActiveReasonNotSupportedWithTopologyAwareScheduling)
-				messages = append(messages, "TAS is not supported for preemption within cluster queue")
-			}
 			if len(c.multiKueueAdmissionChecks) > 0 {
 				reasons = append(reasons, kueue.ClusterQueueActiveReasonNotSupportedWithTopologyAwareScheduling)
 				messages = append(messages, "TAS is not supported with MultiKueue admission check")
@@ -321,7 +317,6 @@ func (c *clusterQueue) isTASViolated() bool {
 		}
 	}
 	return c.HasParent() ||
-		c.Preemption.WithinClusterQueue != kueue.PreemptionPolicyNever ||
 		len(c.multiKueueAdmissionChecks) > 0 ||
 		len(c.provisioningAdmissionChecks) > 0
 }
diff --git a/pkg/cache/clusterqueue_snapshot.go b/pkg/cache/clusterqueue_snapshot.go
index 60aa8fd2c..e9b45bec1 100644
--- a/pkg/cache/clusterqueue_snapshot.go
+++ b/pkg/cache/clusterqueue_snapshot.go
@@ -25,9 +25,11 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/hierarchy"
 	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/resources"
+	utiltas "sigs.k8s.io/kueue/pkg/util/tas"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
@@ -78,6 +80,19 @@ func (c *ClusterQueueSnapshot) removeUsage(frq resources.FlavorResourceQuantitie
 	}
 }
 
+func (c *ClusterQueueSnapshot) updateTASUsage(wi *workload.Info, op usageOp) {
+	if features.Enabled(features.TopologyAwareScheduling) && wi.IsUsingTAS() {
+		for tasFlavor, tasUsage := range wi.TASUsage() {
+			if tasFlvCache := c.TASFlavors[tasFlavor]; tasFlvCache != nil {
+				for _, tr := range tasUsage {
+					domainID := utiltas.DomainID(tr.Values)
+					tasFlvCache.updateTASUsage(domainID, tr.Requests, op)
+				}
+			}
+		}
+	}
+}
+
 func (c *ClusterQueueSnapshot) Fits(frq resources.FlavorResourceQuantities) bool {
 	for fr, q := range frq {
 		if c.Available(fr) < q {
@@ -153,14 +168,15 @@ func (c *ClusterQueueSnapshot) DominantResourceShareWithout(wlReq resources.Flav
 type WorkloadTASRequests map[kueue.ResourceFlavorReference]FlavorTASRequests
 
 func (c *ClusterQueueSnapshot) FindTopologyAssignmentsForWorkload(
-	tasRequestsByFlavor WorkloadTASRequests) TASAssignmentsResult {
+	tasRequestsByFlavor WorkloadTASRequests,
+	simulateEmpty bool) TASAssignmentsResult {
 	result := make(TASAssignmentsResult)
 	for tasFlavor, flavorTASRequests := range tasRequestsByFlavor {
 		// We assume the `tasFlavor` is already in the snapshot as this was
 		// already checked earlier during flavor assignment, and the set of
 		// flavors is immutable in snapshot.
 		tasFlavorCache := c.TASFlavors[tasFlavor]
-		flvResult := tasFlavorCache.FindTopologyAssignmentsForFlavor(flavorTASRequests)
+		flvResult := tasFlavorCache.FindTopologyAssignmentsForFlavor(flavorTASRequests, simulateEmpty)
 		for psName, psAssignment := range flvResult {
 			result[psName] = psAssignment
 		}
diff --git a/pkg/cache/snapshot.go b/pkg/cache/snapshot.go
index fe615a12e..6623dca6c 100644
--- a/pkg/cache/snapshot.go
+++ b/pkg/cache/snapshot.go
@@ -45,6 +45,7 @@ func (s *Snapshot) RemoveWorkload(wl *workload.Info) {
 	cq := s.ClusterQueues[wl.ClusterQueue]
 	delete(cq.Workloads, workload.Key(wl.Obj))
 	cq.removeUsage(wl.FlavorResourceUsage())
+	cq.updateTASUsage(wl, subtract)
 }
 
 // AddWorkload adds a workload from its corresponding ClusterQueue and
@@ -53,6 +54,7 @@ func (s *Snapshot) AddWorkload(wl *workload.Info) {
 	cq := s.ClusterQueues[wl.ClusterQueue]
 	cq.Workloads[workload.Key(wl.Obj)] = wl
 	cq.AddUsage(wl.FlavorResourceUsage())
+	cq.updateTASUsage(wl, add)
 }
 
 func (s *Snapshot) Log(log logr.Logger) {
diff --git a/pkg/cache/tas_flavor_snapshot.go b/pkg/cache/tas_flavor_snapshot.go
index 909c888b1..9757c2a9d 100644
--- a/pkg/cache/tas_flavor_snapshot.go
+++ b/pkg/cache/tas_flavor_snapshot.go
@@ -212,6 +212,14 @@ func (s *TASFlavorSnapshot) addNonTASUsage(domainID utiltas.TopologyDomainID, us
 	s.leaves[domainID].freeCapacity.Sub(resources.Requests{corev1.ResourcePods: 1})
 }
 
+func (s *TASFlavorSnapshot) updateTASUsage(domainID utiltas.TopologyDomainID, usage resources.Requests, op usageOp) {
+	if op == add {
+		s.addTASUsage(domainID, usage)
+	} else {
+		s.removeTASUsage(domainID, usage)
+	}
+}
+
 func (s *TASFlavorSnapshot) addTASUsage(domainID utiltas.TopologyDomainID, usage resources.Requests) {
 	if s.leaves[domainID] == nil {
 		// this can happen if there is an admitted workload for which the
@@ -226,6 +234,13 @@ func (s *TASFlavorSnapshot) addTASUsage(domainID utiltas.TopologyDomainID, usage
 	s.leaves[domainID].tasUsage.Add(usage)
 }
 
+func (s *TASFlavorSnapshot) removeTASUsage(domainID utiltas.TopologyDomainID, usage resources.Requests) {
+	if s.leaves[domainID].tasUsage == nil {
+		s.leaves[domainID].tasUsage = resources.Requests{}
+	}
+	s.leaves[domainID].tasUsage.Sub(usage)
+}
+
 type TASPodSetRequests struct {
 	PodSet            *kueue.PodSet
 	SinglePodRequests resources.Requests
@@ -261,11 +276,15 @@ type tasPodSetAssignmentResult struct {
 
 type FlavorTASRequests []TASPodSetRequests
 
-func (s *TASFlavorSnapshot) FindTopologyAssignmentsForFlavor(flavorTASRequests FlavorTASRequests) TASAssignmentsResult {
+// FindTopologyAssignmentsForFlavor returns TAS assignment, if possible, for all
+// the TAS requests in the flavor handled by the snapshot.
+// The simulateEmpty parameter allows to look for the assignment under the
+// assumption that all TAS workloads are preempted.
+func (s *TASFlavorSnapshot) FindTopologyAssignmentsForFlavor(flavorTASRequests FlavorTASRequests, simulateEmpty bool) TASAssignmentsResult {
 	result := make(map[string]tasPodSetAssignmentResult)
 	assumedUsage := make(map[utiltas.TopologyDomainID]resources.Requests)
 	for _, tr := range flavorTASRequests {
-		assignment, reason := s.findTopologyAssignment(tr, assumedUsage)
+		assignment, reason := s.findTopologyAssignment(tr, assumedUsage, simulateEmpty)
 		result[tr.PodSet.Name] = tasPodSetAssignmentResult{TopologyAssignment: assignment, FailureReason: reason}
 		if reason != "" {
 			return result
@@ -297,7 +316,8 @@ func (s *TASFlavorSnapshot) FindTopologyAssignmentsForFlavor(flavorTASRequests F
 //	c) build the assignment for the lowest level in the hierarchy
 func (s *TASFlavorSnapshot) findTopologyAssignment(
 	tasPodSetRequests TASPodSetRequests,
-	assumedUsage map[utiltas.TopologyDomainID]resources.Requests) (*kueue.TopologyAssignment, string) {
+	assumedUsage map[utiltas.TopologyDomainID]resources.Requests,
+	simulateEmpty bool) (*kueue.TopologyAssignment, string) {
 	topologyRequest := tasPodSetRequests.PodSet.TopologyRequest
 	requests := tasPodSetRequests.SinglePodRequests
 	requests.Add(resources.Requests{corev1.ResourcePods: 1})
@@ -313,7 +333,7 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 		return nil, fmt.Sprintf("no requested topology level: %s", *key)
 	}
 	// phase 1 - determine the number of pods which can fit in each topology domain
-	s.fillInCounts(requests, assumedUsage, append(podSetTolerations, s.tolerations...))
+	s.fillInCounts(requests, assumedUsage, simulateEmpty, append(podSetTolerations, s.tolerations...))
 
 	// phase 2a: determine the level at which the assignment is done along with
 	// the domains which can accommodate all pods
@@ -487,6 +507,7 @@ func (s *TASFlavorSnapshot) sortedDomains(domains []*domain) []*domain {
 
 func (s *TASFlavorSnapshot) fillInCounts(requests resources.Requests,
 	assumedUsage map[utiltas.TopologyDomainID]resources.Requests,
+	simulateEmpty bool,
 	tolerations []corev1.Toleration) {
 	for _, domain := range s.domains {
 		// cleanup the state in case some remaining values are present from computing
@@ -502,7 +523,9 @@ func (s *TASFlavorSnapshot) fillInCounts(requests resources.Requests,
 			continue
 		}
 		remainingCapacity := leaf.freeCapacity.Clone()
-		remainingCapacity.Sub(leaf.tasUsage)
+		if !simulateEmpty {
+			remainingCapacity.Sub(leaf.tasUsage)
+		}
 		if leafAssumedUsage, found := assumedUsage[leaf.domain.id]; found {
 			remainingCapacity.Sub(leafAssumedUsage)
 		}
diff --git a/pkg/scheduler/flavorassigner/flavorassigner.go b/pkg/scheduler/flavorassigner/flavorassigner.go
index d2a9ee037..248e1da0c 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner.go
@@ -204,6 +204,12 @@ func (psa *PodSetAssignment) RepresentativeMode() FlavorAssignmentMode {
 	return mode
 }
 
+func (psa *PodSetAssignment) updateMode(newMode FlavorAssignmentMode) {
+	for _, flvAssignment := range psa.Flavors {
+		flvAssignment.Mode = newMode
+	}
+}
+
 func (psa *PodSetAssignment) reason(reason string) {
 	if psa.Status == nil {
 		psa.Status = &Status{}
@@ -395,22 +401,34 @@ func (a *FlavorAssigner) assignFlavors(log logr.Logger, counts []int32) Assignme
 		return assignment
 	}
 
-	if features.Enabled(features.TopologyAwareScheduling) && assignment.RepresentativeMode() == Fit {
+	if features.Enabled(features.TopologyAwareScheduling) {
 		tasRequests := assignment.WorkloadsTopologyRequests(a.wl, a.cq)
-		result := a.cq.FindTopologyAssignmentsForWorkload(tasRequests)
-		if failure := result.Failure(); failure != nil {
-			// There is at least one PodSet which does not fit
-			psAssignment := assignment.podSetAssignmentByName(failure.PodSetName)
-			psAssignment.reason(failure.Reason)
-			// clear flavors to make the PodSet assignment as NoFit
-			psAssignment.Flavors = nil
-			// update the representative mode accordingly
-			assignment.representativeMode = ptr.To(NoFit)
-		} else {
-			// All PodSets fit, we just update the TopologyAssignments
-			for psName, psResult := range result {
-				psAssignment := assignment.podSetAssignmentByName(psName)
-				psAssignment.TopologyAssignment = psResult.TopologyAssignment
+		if assignment.RepresentativeMode() == Fit {
+			result := a.cq.FindTopologyAssignmentsForWorkload(tasRequests, false)
+			if failure := result.Failure(); failure != nil {
+				// There is at least one PodSet which does not fit
+				psAssignment := assignment.podSetAssignmentByName(failure.PodSetName)
+				psAssignment.reason(failure.Reason)
+				// update the mode for all flavors and the representative mode
+				psAssignment.updateMode(Preempt)
+				assignment.representativeMode = ptr.To(Preempt)
+			} else {
+				// All PodSets fit, we just update the TopologyAssignments
+				for psName, psResult := range result {
+					psAssignment := assignment.podSetAssignmentByName(psName)
+					psAssignment.TopologyAssignment = psResult.TopologyAssignment
+				}
+			}
+		}
+		if assignment.RepresentativeMode() == Preempt {
+			result := a.cq.FindTopologyAssignmentsForWorkload(tasRequests, true)
+			if failure := result.Failure(); failure != nil {
+				// There is at least one PodSet which does not fit even if
+				// all workloads are preempted.
+				psAssignment := assignment.podSetAssignmentByName(failure.PodSetName)
+				// update the mode for all flavors and the representative mode
+				psAssignment.updateMode(NoFit)
+				assignment.representativeMode = ptr.To(NoFit)
 			}
 		}
 	}
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 4df37e858..688e7050a 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -69,6 +69,7 @@ type preemptionCtx struct {
 	preemptorCQ       *cache.ClusterQueueSnapshot
 	snapshot          *cache.Snapshot
 	requests          resources.FlavorResourceQuantities
+	tasRequests       cache.WorkloadTASRequests
 	frsNeedPreemption sets.Set[resources.FlavorResource]
 }
 
@@ -123,12 +124,15 @@ type Target struct {
 // GetTargets returns the list of workloads that should be evicted in
 // order to make room for wl.
 func (p *Preemptor) GetTargets(log logr.Logger, wl workload.Info, assignment flavorassigner.Assignment, snapshot *cache.Snapshot) []*Target {
+	cq := snapshot.ClusterQueues[wl.ClusterQueue]
+	tasRequests := assignment.WorkloadsTopologyRequests(&wl, cq)
 	return p.getTargets(&preemptionCtx{
 		log:               log,
 		preemptor:         wl,
-		preemptorCQ:       snapshot.ClusterQueues[wl.ClusterQueue],
+		preemptorCQ:       cq,
 		snapshot:          snapshot,
 		requests:          assignment.TotalRequestsFor(&wl),
+		tasRequests:       tasRequests,
 		frsNeedPreemption: flavorResourcesNeedPreemption(assignment),
 	})
 }
@@ -579,7 +583,8 @@ func workloadFits(preemptionCtx *preemptionCtx, allowBorrowing bool) bool {
 			return false
 		}
 	}
-	return true
+	tasResult := preemptionCtx.preemptorCQ.FindTopologyAssignmentsForWorkload(preemptionCtx.tasRequests, false)
+	return tasResult.Failure() == nil
 }
 
 func queueUnderNominalInResourcesNeedingPreemption(preemptionCtx *preemptionCtx) bool {
diff --git a/site/content/en/docs/concepts/multikueue.md b/site/content/en/docs/concepts/multikueue.md
index c606df5af..d70c9bf4e 100644
--- a/site/content/en/docs/concepts/multikueue.md
+++ b/site/content/en/docs/concepts/multikueue.md
@@ -76,6 +76,21 @@ The supported version of the Kubeflow Training Operator is v1.7.0, or a newer ve
 The Management cluster should only install the CRDs and not the package itself. 
 On the other hand, the Worker cluster should install the full kubeflow operator.
 
+## Plain Pods
+
+MultiKueue supports the remote creation and management of Plain Pods.
+
+## Deployments
+
+MultiKueue supports the remote creation and management of Deployment replica Pods.
+
+Known Limitations:
+- When creating a Deployment in environments with more than 1 worker cluster it is possible that replicas are scheduled in different clusters.
+
+{{% alert title="Note" color="primary" %}}
+Follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin) to learn how to enable and configure the `pod` integration which is required for enabling the `deployment` integration.
+{{% /alert %}}
+
 ## Submitting Jobs
 In a [configured MultiKueue environment](/docs/tasks/manage/setup_multikueue), you can submit any MultiKueue supported job to the Manager cluster, targeting a ClusterQueue configured for Multikueue.
 Kueue delegates the job to the configured worker clusters without any additional configuration changes.
diff --git a/site/content/en/docs/tasks/run/deployment.md b/site/content/en/docs/tasks/run/deployment.md
index 2ff88aa16..2556ab299 100644
--- a/site/content/en/docs/tasks/run/deployment.md
+++ b/site/content/en/docs/tasks/run/deployment.md
@@ -24,7 +24,7 @@ For more information, see [Kueue's overview](/docs/overview).
 1. Learn how to [install Kueue with a custom manager configuration](/docs/installation/#install-a-custom-configured-released-version).
 
 2. Follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin)
-to learn how to enable and configure the `v1/pod` integration.
+to learn how to enable and configure the `pod` integration.
 
 3. Check [Administer cluster quotas](/docs/tasks/manage/administer_cluster_quotas) for details on the initial Kueue setup.
 
diff --git a/site/content/en/docs/tasks/run/external_workloads/argo_workflow.md b/site/content/en/docs/tasks/run/external_workloads/argo_workflow.md
index fe7de8039..b31c83bdb 100644
--- a/site/content/en/docs/tasks/run/external_workloads/argo_workflow.md
+++ b/site/content/en/docs/tasks/run/external_workloads/argo_workflow.md
@@ -19,7 +19,7 @@ but you can take advantage of the ability for Kueue to [manage plain pods](/docs
 1. Learn how to [install Kueue with a custom manager configuration](/docs/installation/#install-a-custom-configured-released-version).
 
 2. Follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin)
-to learn how to enable and configure the `v1/pod` integration.
+to learn how to enable and configure the `pod` integration.
 
 3. Install [Argo Workflows](https://argo-workflows.readthedocs.io/en/latest/installation/#installation)
 
diff --git a/site/content/en/docs/tasks/run/external_workloads/tektoncd.md b/site/content/en/docs/tasks/run/external_workloads/tektoncd.md
index c3ec608a2..f309d42d9 100644
--- a/site/content/en/docs/tasks/run/external_workloads/tektoncd.md
+++ b/site/content/en/docs/tasks/run/external_workloads/tektoncd.md
@@ -17,7 +17,7 @@ We demonstrate how to support scheduling Tekton Pipelines Tasks in Kueue based o
 
 1. Learn how to [install Kueue with a custom manager configuration](/docs/installation/#install-a-custom-configured-released-version).
 
-1. Follow the steps in [Run Plain Pods](docs/tasks/run/plain_pods/#before-you-begin) to learn how to enable and configure the `v1/pod` integration.
+1. Follow the steps in [Run Plain Pods](docs/tasks/run/plain_pods/#before-you-begin) to learn how to enable and configure the `pod` integration.
 
 1. Check [Administrator cluster quotas](/docs/tasks/manage/administer_cluster_quotas/) for details on the initial Kueue step.
 
diff --git a/site/content/en/docs/tasks/run/multikueue/deployment.md b/site/content/en/docs/tasks/run/multikueue/deployment.md
new file mode 100644
index 000000000..c3a166359
--- /dev/null
+++ b/site/content/en/docs/tasks/run/multikueue/deployment.md
@@ -0,0 +1,29 @@
+---
+title: "Run Deployment in Multi-Cluster"
+linkTitle: "Deployment"
+weight: 2
+date: 2025-02-17
+description: >
+  Run a MultiKueue scheduled Deployment.
+---
+
+## Before you begin
+
+1. Check the [MultiKueue installation guide](/docs/tasks/manage/setup_multikueue) on how to properly setup MultiKueue clusters.
+
+2. Follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin)
+to learn how to enable and configure the `pod` integration which is required for enabling the `deployment` integration.
+
+Deployments receive live status updates through the status from the remote Pods created on the worker cluster.
+
+{{< feature-state state="beta" for_version="v0.11.0" >}}
+
+{{% alert title="Note" color="primary" %}}
+In this current implementation, when Deployments are created in environments with multiple worker clusters, Pods are allocated to any worker.
+{{% /alert %}}
+
+## Example
+
+Once the setup is complete you can test it by running the example below:
+
+{{< include "examples/serving-workloads/sample-deployment.yaml" "yaml" >}}
diff --git a/site/content/en/docs/tasks/run/multikueue/plain_pods.md b/site/content/en/docs/tasks/run/multikueue/plain_pods.md
new file mode 100644
index 000000000..becdde494
--- /dev/null
+++ b/site/content/en/docs/tasks/run/multikueue/plain_pods.md
@@ -0,0 +1,25 @@
+---
+title: "Run Plain Pod in Multi-Cluster"
+linkTitle: "Plain Pod"
+weight: 2
+date: 2025-02-17
+description: >
+  Run a MultiKueue scheduled Plain Pod.
+---
+
+## Before you begin
+
+1. Check the [MultiKueue installation guide](/docs/tasks/manage/setup_multikueue) on how to properly setup MultiKueue clusters.
+
+2. Follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin)
+to learn how to enable and configure the `pod` integration.
+
+Pods created on the manager cluster are automatically gated and receive live status updates from their remote counterparts
+
+{{< feature-state state="beta" for_version="v0.11.0" >}}
+
+## Example
+
+Once the setup is complete you can test it by running the example below:
+
+{{< include "examples/pods-kueue/kueue-pod.yaml" "yaml" >}}
diff --git a/site/content/en/docs/tasks/run/plain_pods.md b/site/content/en/docs/tasks/run/plain_pods.md
index ae485917d..cf8a0231a 100644
--- a/site/content/en/docs/tasks/run/plain_pods.md
+++ b/site/content/en/docs/tasks/run/plain_pods.md
@@ -16,7 +16,7 @@ This guide is for [batch users](/docs/tasks#batch-user) that have a basic unders
 
 ## Before you begin
 
-1. By default, the integration for `v1/pod` is not enabled.
+1. By default, the integration for `pod` is not enabled.
    Learn how to [install Kueue with a custom manager configuration](/docs/installation/#install-a-custom-configured-released-version)
    and enable the `pod` integration.
 
diff --git a/site/content/en/docs/tasks/run/statefulset.md b/site/content/en/docs/tasks/run/statefulset.md
index 1c15bfb4f..7261469a9 100644
--- a/site/content/en/docs/tasks/run/statefulset.md
+++ b/site/content/en/docs/tasks/run/statefulset.md
@@ -31,7 +31,7 @@ For more information, see [Kueue's overview](/docs/overview).
       - "statefulset"
    ```
    Also, follow steps in [Run Plain Pods](/docs/tasks/run/plain_pods/#before-you-begin)
-   to learn how to enable and configure the `v1/pod` integration.
+   to learn how to enable and configure the `pod` integration.
 
 3. Check [Administer cluster quotas](/docs/tasks/manage/administer_cluster_quotas) for details on the initial Kueue setup.
 



Test patch:
diff --git a/pkg/cache/clusterqueue_test.go b/pkg/cache/clusterqueue_test.go
index 0ce9c4fad..761cbdcfe 100644
--- a/pkg/cache/clusterqueue_test.go
+++ b/pkg/cache/clusterqueue_test.go
@@ -679,8 +679,8 @@ func TestClusterQueueReadinessWithTAS(t *testing.T) {
 					WhenCanPreempt: kueue.Preempt,
 				}).
 				Obj(),
-			wantReason:  kueue.ClusterQueueActiveReasonNotSupportedWithTopologyAwareScheduling,
-			wantMessage: "Can't admit new workloads: TAS is not supported for preemption within cluster queue.",
+			wantReason:  kueue.ClusterQueueActiveReasonReady,
+			wantMessage: "Can admit new workloads",
 		},
 		{
 			name: "TAS do not support MultiKueue AdmissionCheck",
diff --git a/pkg/cache/tas_cache_test.go b/pkg/cache/tas_cache_test.go
index 1d7e31324..2861005d1 100644
--- a/pkg/cache/tas_cache_test.go
+++ b/pkg/cache/tas_cache_test.go
@@ -1437,7 +1437,7 @@ func TestFindTopologyAssignment(t *testing.T) {
 				wantMainPodSetResult.TopologyAssignment = tc.wantAssignment
 			}
 			wantResult["main"] = wantMainPodSetResult
-			gotResult := snapshot.FindTopologyAssignmentsForFlavor(flavorTASRequests)
+			gotResult := snapshot.FindTopologyAssignmentsForFlavor(flavorTASRequests, false)
 			if diff := cmp.Diff(wantResult, gotResult); diff != "" {
 				t.Errorf("unexpected topology assignment (-want,+got): %s", diff)
 			}
diff --git a/pkg/controller/jobframework/base_webhook_test.go b/pkg/controller/jobframework/base_webhook_test.go
index 218b00c03..d3fbd61c0 100644
--- a/pkg/controller/jobframework/base_webhook_test.go
+++ b/pkg/controller/jobframework/base_webhook_test.go
@@ -148,7 +148,7 @@ func TestBaseWebhookDefault(t *testing.T) {
 		manageJobsWithoutQueueName bool
 		localQueueDefaulting       bool
 		defaultLqExist             bool
-		multiQueue                 bool
+		enableMultiKueue           bool
 		job                        *batchv1.Job
 		want                       *batchv1.Job
 	}{
@@ -222,7 +222,7 @@ func TestBaseWebhookDefault(t *testing.T) {
 				Queue("multikueue").
 				ManagedBy(kueue.MultiKueueControllerName).
 				Obj(),
-			multiQueue: true,
+			enableMultiKueue: true,
 		},
 		"ManagedByDefaulting, targeting multikueue local queue but already managaed by someone else": {
 			job: utiljob.MakeJob("job", "default").
@@ -233,7 +233,7 @@ func TestBaseWebhookDefault(t *testing.T) {
 				Queue("multikueue").
 				ManagedBy("someone-else").
 				Obj(),
-			multiQueue: true,
+			enableMultiKueue: true,
 		},
 		"ManagedByDefaulting, targeting non-multikueue local queue": {
 			job: utiljob.MakeJob("job", "default").
@@ -242,14 +242,14 @@ func TestBaseWebhookDefault(t *testing.T) {
 			want: utiljob.MakeJob("job", "default").
 				Queue("queue").
 				Obj(),
-			multiQueue: true,
+			enableMultiKueue: true,
 		},
 	}
 	for name, tc := range testcases {
 		t.Run(name, func(t *testing.T) {
 			ctx, _ := utiltesting.ContextWithLog(t)
 			features.SetFeatureGateDuringTest(t, features.LocalQueueDefaulting, tc.localQueueDefaulting)
-			features.SetFeatureGateDuringTest(t, features.MultiKueue, tc.multiQueue)
+			features.SetFeatureGateDuringTest(t, features.MultiKueue, tc.enableMultiKueue)
 			clientBuilder := utiltesting.NewClientBuilder().
 				WithObjects(
 					&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}},
@@ -263,7 +263,7 @@ func TestBaseWebhookDefault(t *testing.T) {
 					t.Fatalf("failed to create default local queue: %s", err)
 				}
 			}
-			if tc.multiQueue {
+			if tc.enableMultiKueue {
 				if err := queueManager.AddLocalQueue(ctx, utiltesting.MakeLocalQueue("multikueue", "default").
 					ClusterQueue("cluster-queue").Obj()); err != nil {
 					t.Fatalf("failed to create default local queue: %s", err)
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 79e6a2c59..56861444b 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -5105,3 +5105,510 @@ func TestScheduleForTAS(t *testing.T) {
 		})
 	}
 }
+
+func TestScheduleForTASPreemption(t *testing.T) {
+	defaultSingleNode := []corev1.Node{
+		*testingnode.MakeNode("x1").
+			Label("tas-node", "true").
+			Label(corev1.LabelHostname, "x1").
+			StatusAllocatable(corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("5"),
+				corev1.ResourceMemory: resource.MustParse("5Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
+			}).
+			Ready().
+			Obj(),
+	}
+	defaultTwoNodes := []corev1.Node{
+		*testingnode.MakeNode("x1").
+			Label("tas-node", "true").
+			Label(corev1.LabelHostname, "x1").
+			StatusAllocatable(corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("5"),
+				corev1.ResourceMemory: resource.MustParse("5Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
+			}).
+			Ready().
+			Obj(),
+		*testingnode.MakeNode("y1").
+			Label("tas-node", "true").
+			Label(corev1.LabelHostname, "y1").
+			StatusAllocatable(corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("5"),
+				corev1.ResourceMemory: resource.MustParse("5Gi"),
+				corev1.ResourcePods:   resource.MustParse("10"),
+			}).
+			Ready().
+			Obj(),
+	}
+	defaultSingleLevelTopology := *utiltesting.MakeDefaultOneLevelTopology("tas-single-level")
+	defaultTASFlavor := *utiltesting.MakeResourceFlavor("tas-default").
+		NodeLabel("tas-node", "true").
+		TopologyName("tas-single-level").
+		Obj()
+	defaultClusterQueueWithPreemption := *utiltesting.MakeClusterQueue("tas-main").
+		Preemption(kueue.ClusterQueuePreemption{WithinClusterQueue: kueue.PreemptionPolicyLowerPriority}).
+		ResourceGroup(*utiltesting.MakeFlavorQuotas("tas-default").
+			Resource(corev1.ResourceCPU, "50").
+			Resource(corev1.ResourceMemory, "50Gi").Obj()).
+		Obj()
+	queues := []kueue.LocalQueue{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "default",
+				Name:      "tas-main",
+			},
+			Spec: kueue.LocalQueueSpec{
+				ClusterQueue: "tas-main",
+			},
+		},
+	}
+	cases := map[string]struct {
+		nodes           []corev1.Node
+		pods            []corev1.Pod
+		topologies      []kueuealpha.Topology
+		resourceFlavors []kueue.ResourceFlavor
+		clusterQueues   []kueue.ClusterQueue
+		workloads       []kueue.Workload
+
+		// wantNewAssignments is a summary of all new admissions in the cache after this cycle.
+		wantNewAssignments map[string]kueue.Admission
+		// wantLeft is the workload keys that are left in the queues after this cycle.
+		wantLeft map[string][]string
+		// wantInadmissibleLeft is the workload keys that are left in the inadmissible state after this cycle.
+		wantInadmissibleLeft map[string][]string
+		// wantPreempted is the keys of the workloads that get preempted in the scheduling cycle.
+		wantPreempted sets.Set[string]
+		// wantEvents asserts on the events, the comparison options are passed by eventCmpOpts
+		wantEvents []utiltesting.EventRecord
+		// eventCmpOpts are the comparison options for the events
+		eventCmpOpts []cmp.Option
+	}{
+		"only low priority workload is preempted": {
+			// This test case demonstrates the baseline scenario where there
+			// is only one low-priority workload and it gets preempted.
+			nodes:           defaultSingleNode,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueWithPreemption},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo", "default").
+					Queue("tas-main").
+					Priority(3).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "2").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("low-priority-admitted", "default").
+					Queue("tas-main").
+					Priority(1).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "5").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "5").
+						Obj()).
+					Obj(),
+			},
+			wantPreempted: sets.New("default/low-priority-admitted"),
+			wantLeft: map[string][]string{
+				"tas-main": {"default/foo"},
+			},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "low-priority-admitted"},
+					EventType: "Normal",
+					Reason:    "Preempted",
+					Message:   "Preempted to accommodate a workload (UID: ) due to prioritization in the ClusterQueue",
+				},
+				{
+					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). Pending the preemption of 1 workload(s)`,
+				},
+			},
+		},
+		"low priority workload is preempted, mid-priority workload survives": {
+			// This test case demonstrates the targets are selected according
+			// to priorities, similarly as for regular preemption.
+			nodes:           defaultSingleNode,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueWithPreemption},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo", "default").
+					Queue("tas-main").
+					Priority(3).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "2").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("mid-priority-admitted", "default").
+					Queue("tas-main").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "2").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "2").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("low-priority-admitted", "default").
+					Queue("tas-main").
+					Priority(1).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "2").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "2").
+						Obj()).
+					Obj(),
+			},
+			wantPreempted: sets.New("default/low-priority-admitted"),
+			wantLeft: map[string][]string{
+				"tas-main": {"default/foo"},
+			},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "low-priority-admitted"},
+					EventType: "Normal",
+					Reason:    "Preempted",
+					Message:   "Preempted to accommodate a workload (UID: ) due to prioritization in the ClusterQueue",
+				},
+				{
+					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). Pending the preemption of 1 workload(s)`,
+				},
+			},
+		},
+		"low priority workload is preempted even though there is enough capacity, but fragmented": {
+			// In this test we fill in two nodes 4/5 which leaves 2 units empty.
+			// It would be enough to schedule both pods of wl3, one pod per
+			// node, but the workload requires to run on a single node, thus
+			// the lower priority workload is chosen as target.
+			nodes:           defaultTwoNodes,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueWithPreemption},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo", "default").
+					Queue("tas-main").
+					Priority(3).
+					PodSets(*utiltesting.MakePodSet("one", 2).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("mid-priority-admitted", "default").
+					Queue("tas-main").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "4").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "4").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("low-priority-admitted", "default").
+					Queue("tas-main").
+					Priority(1).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "4").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"y1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "4").
+						Obj()).
+					Obj(),
+			},
+			wantPreempted: sets.New("default/low-priority-admitted"),
+			wantLeft: map[string][]string{
+				"tas-main": {"default/foo"},
+			},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "low-priority-admitted"},
+					EventType: "Normal",
+					Reason:    "Preempted",
+					Message:   "Preempted to accommodate a workload (UID: ) due to prioritization in the ClusterQueue",
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "foo"},
+					EventType: "Warning",
+					Reason:    "Pending",
+					Message:   `couldn't assign flavors to pod set one: topology "tas-single-level" allows to fit only 1 out of 2 pod(s). Pending the preemption of 1 workload(s)`,
+				},
+			},
+		},
+		"workload with equal priority awaits for other workloads to complete": {
+			// In this test case the waiting workload cannot preempt the running
+			// workload as they are both with the same priority. Still, the
+			// waiting workload does not let the second workload in the queue
+			// to get in, as it is awaiting for the running workloads to
+			// complete.
+			nodes:           defaultSingleNode,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueueWithPreemption},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("low-priority-which-would-fit", "default").
+					Queue("tas-main").
+					Priority(1).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("mid-priority-waiting", "default").
+					Queue("tas-main").
+					Priority(2).
+					PodSets(*utiltesting.MakePodSet("one", 2).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+				*utiltesting.MakeWorkload("mid-priority-admitted", "default").
+					Queue("tas-main").
+					Priority(2).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "4").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"x1",
+										},
+									},
+								},
+							}).Obj(),
+					).
+					Admitted(true).
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						RequiredTopologyRequest(corev1.LabelHostname).
+						Request(corev1.ResourceCPU, "4").
+						Obj()).
+					Obj(),
+			},
+			wantLeft: map[string][]string{
+				"tas-main": {"default/low-priority-which-would-fit"},
+			},
+			wantInadmissibleLeft: map[string][]string{
+				"tas-main": {"default/mid-priority-waiting"},
+			},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "mid-priority-waiting"},
+					EventType: "Warning",
+					Reason:    "Pending",
+					Message:   `couldn't assign flavors to pod set one: topology "tas-single-level" allows to fit only 1 out of 2 pod(s)`,
+				},
+			},
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			features.SetFeatureGateDuringTest(t, features.TopologyAwareScheduling, true)
+			ctx, _ := utiltesting.ContextWithLog(t)
+
+			clientBuilder := utiltesting.NewClientBuilder().
+				WithLists(
+					&kueue.WorkloadList{Items: tc.workloads},
+					&kueuealpha.TopologyList{Items: tc.topologies},
+					&corev1.PodList{Items: tc.pods},
+					&corev1.NodeList{Items: tc.nodes},
+					&kueue.LocalQueueList{Items: queues}).
+				WithObjects(
+					&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}},
+				)
+			_ = tasindexer.SetupIndexes(ctx, utiltesting.AsIndexer(clientBuilder))
+			cl := clientBuilder.Build()
+			recorder := &utiltesting.EventRecorder{}
+			cqCache := cache.New(cl)
+			qManager := queue.NewManager(cl, cqCache)
+			topologyByName := slices.ToMap(tc.topologies, func(i int) (kueue.TopologyReference, kueuealpha.Topology) {
+				return kueue.TopologyReference(tc.topologies[i].Name), tc.topologies[i]
+			})
+			for _, flavor := range tc.resourceFlavors {
+				cqCache.AddOrUpdateResourceFlavor(&flavor)
+				if flavor.Spec.TopologyName != nil {
+					t := topologyByName[*flavor.Spec.TopologyName]
+					tasCache := cqCache.TASCache()
+					levels := utiltas.Levels(&t)
+					tasFlavorCache := tasCache.NewTASFlavorCache(*flavor.Spec.TopologyName, levels, flavor.Spec.NodeLabels, flavor.Spec.Tolerations)
+					tasCache.Set(kueue.ResourceFlavorReference(flavor.Name), tasFlavorCache)
+				}
+			}
+			for _, cq := range tc.clusterQueues {
+				if err := cqCache.AddClusterQueue(ctx, &cq); err != nil {
+					t.Fatalf("Inserting clusterQueue %s in cache: %v", cq.Name, err)
+				}
+				if err := qManager.AddClusterQueue(ctx, &cq); err != nil {
+					t.Fatalf("Inserting clusterQueue %s in manager: %v", cq.Name, err)
+				}
+				if err := cl.Create(ctx, &cq); err != nil {
+					t.Fatalf("couldn't create the cluster queue: %v", err)
+				}
+			}
+			for _, q := range queues {
+				if err := qManager.AddLocalQueue(ctx, &q); err != nil {
+					t.Fatalf("Inserting queue %s/%s in manager: %v", q.Namespace, q.Name, err)
+				}
+			}
+			initiallyAdmittedWorkloads := sets.New[string]()
+			for _, w := range tc.workloads {
+				if workload.IsAdmitted(&w) {
+					initiallyAdmittedWorkloads.Insert(workload.Key(&w))
+				}
+			}
+			scheduler := New(qManager, cqCache, cl, recorder)
+			gotScheduled := make([]string, 0)
+			var mu sync.Mutex
+			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
+				mu.Lock()
+				gotScheduled = append(gotScheduled, workload.Key(w))
+				mu.Unlock()
+				return nil
+			}
+			wg := sync.WaitGroup{}
+			scheduler.setAdmissionRoutineWrapper(routine.NewWrapper(
+				func() { wg.Add(1) },
+				func() { wg.Done() },
+			))
+
+			gotPreempted := sets.New[string]()
+			scheduler.preemptor.OverrideApply(func(_ context.Context, w *kueue.Workload, _, _ string) error {
+				mu.Lock()
+				gotPreempted.Insert(workload.Key(w))
+				mu.Unlock()
+				return nil
+			})
+
+			ctx, cancel := context.WithTimeout(ctx, queueingTimeout)
+			go qManager.CleanUpOnContext(ctx)
+			defer cancel()
+
+			scheduler.schedule(ctx)
+			wg.Wait()
+			snapshot, err := cqCache.Snapshot(ctx)
+			if err != nil {
+				t.Fatalf("unexpected error while building snapshot: %v", err)
+			}
+			if diff := cmp.Diff(tc.wantPreempted, gotPreempted); diff != "" {
+				t.Errorf("Unexpected preemptions (-want,+got):\n%s", diff)
+			}
+			gotAssignments := make(map[string]kueue.Admission)
+			for cqName, c := range snapshot.ClusterQueues {
+				for name, w := range c.Workloads {
+					if initiallyAdmittedWorkloads.Has(workload.Key(w.Obj)) {
+						continue
+					}
+					switch {
+					case !workload.HasQuotaReservation(w.Obj):
+						t.Fatalf("Workload %s is not admitted by a clusterQueue, but it is found as member of clusterQueue %s in the cache", name, cqName)
+					case string(w.Obj.Status.Admission.ClusterQueue) != cqName:
+						t.Fatalf("Workload %s is admitted by clusterQueue %s, but it is found as member of clusterQueue %s in the cache", name, w.Obj.Status.Admission.ClusterQueue, cqName)
+					default:
+						gotAssignments[name] = *w.Obj.Status.Admission
+					}
+				}
+			}
+			if diff := cmp.Diff(tc.wantNewAssignments, gotAssignments, cmpopts.EquateEmpty()); diff != "" {
+				t.Errorf("Unexpected assigned clusterQueues in cache (-want,+got):\n%s", diff)
+			}
+			qDump := qManager.Dump()
+			if diff := cmp.Diff(tc.wantLeft, qDump, cmpDump...); diff != "" {
+				t.Errorf("Unexpected elements left in the queue (-want,+got):\n%s", diff)
+			}
+			qDumpInadmissible := qManager.DumpInadmissible()
+			if diff := cmp.Diff(tc.wantInadmissibleLeft, qDumpInadmissible, cmpDump...); diff != "" {
+				t.Errorf("Unexpected elements left in inadmissible workloads (-want,+got):\n%s", diff)
+			}
+			if diff := cmp.Diff(tc.wantEvents, recorder.RecordedEvents, tc.eventCmpOpts...); diff != "" {
+				t.Errorf("unexpected events (-want/+got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index 0a6b0fd9e..488131769 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -237,29 +237,6 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
-		ginkgo.It("should mark TAS ClusterQueue as inactive if used with preemption", func() {
-			clusterQueue = testing.MakeClusterQueue("cq").
-				ResourceGroup(
-					*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj(),
-				).Preemption(kueue.ClusterQueuePreemption{
-				WithinClusterQueue: kueue.PreemptionPolicyLowerPriority,
-			}).Obj()
-			gomega.Expect(k8sClient.Create(ctx, clusterQueue)).Should(gomega.Succeed())
-
-			gomega.Eventually(func(g gomega.Gomega) {
-				var updatedCq kueue.ClusterQueue
-				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), &updatedCq)).To(gomega.Succeed())
-				g.Expect(updatedCq.Status.Conditions).Should(gomega.BeComparableTo([]metav1.Condition{
-					{
-						Type:    kueue.ClusterQueueActive,
-						Status:  metav1.ConditionFalse,
-						Reason:  "NotSupportedWithTopologyAwareScheduling",
-						Message: `Can't admit new workloads: TAS is not supported for preemption within cluster queue.`,
-					},
-				}, util.IgnoreConditionTimestampsAndObservedGeneration))
-			}, util.Timeout, util.Interval).Should(gomega.Succeed())
-		})
-
 		ginkgo.It("should mark TAS ClusterQueue as inactive if used with MultiKueue", func() {
 			admissionCheck = testing.MakeAdmissionCheck("multikueue").ControllerName(kueue.MultiKueueControllerName).Obj()
 			gomega.Expect(k8sClient.Create(ctx, admissionCheck)).To(gomega.Succeed())
@@ -858,6 +835,127 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 			})
 		})
 
+		ginkgo.When("Preemption is enabled within ClusterQueue", func() {
+			var (
+				nodes []corev1.Node
+			)
+			ginkgo.BeforeEach(func() {
+				nodes = []corev1.Node{
+					*testingnode.MakeNode("x1").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r1").
+						Label(corev1.LabelHostname, "x1").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("5"),
+							corev1.ResourceMemory: resource.MustParse("5Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+					*testingnode.MakeNode("x2").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r2").
+						Label(corev1.LabelHostname, "x2").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("5"),
+							corev1.ResourceMemory: resource.MustParse("5Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+				}
+				util.CreateNodes(ctx, k8sClient, nodes)
+
+				topology = testing.MakeDefaultThreeLevelTopology("default")
+				gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
+
+				tasFlavor = testing.MakeResourceFlavor("tas-flavor").
+					NodeLabel("node-group", "tas").
+					TopologyName("default").Obj()
+				gomega.Expect(k8sClient.Create(ctx, tasFlavor)).Should(gomega.Succeed())
+
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
+					Preemption(kueue.ClusterQueuePreemption{
+						WithinClusterQueue: kueue.PreemptionPolicyLowerPriority,
+					}).
+					// We set quota above what is physically available to hit
+					// the topology requirements rather quota.
+					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).
+						Resource(corev1.ResourceCPU, "20").
+						Resource(corev1.ResourceMemory, "20Gi").Obj()).
+					Obj()
+				gomega.Expect(k8sClient.Create(ctx, clusterQueue)).Should(gomega.Succeed())
+				util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQueue)
+
+				localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+				gomega.Expect(k8sClient.Create(ctx, localQueue)).Should(gomega.Succeed())
+			})
+
+			ginkgo.AfterEach(func() {
+				gomega.Expect(util.DeleteAllJobsInNamespace(ctx, k8sClient, ns)).Should(gomega.Succeed())
+				gomega.Expect(util.DeleteWorkloadsInNamespace(ctx, k8sClient, ns)).Should(gomega.Succeed())
+				gomega.Expect(util.DeleteObject(ctx, k8sClient, localQueue)).Should(gomega.Succeed())
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
+				for _, node := range nodes {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, &node, true)
+				}
+			})
+
+			ginkgo.It("should preempt the low and mid priority workloads to fit the high-priority workload", func() {
+				var wl1, wl2, wl3 *kueue.Workload
+				ginkgo.By("creating a low priority workload which can fit", func() {
+					wl1 = testing.MakeWorkload("wl1", ns.Name).
+						Priority(1).
+						PodSets(*testing.MakePodSet("worker", 1).
+							PreferredTopologyRequest(testing.DefaultBlockTopologyLevel).
+							Obj()).
+						Queue(localQueue.Name).Request(corev1.ResourceCPU, "5").Obj()
+					gomega.Expect(k8sClient.Create(ctx, wl1)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the workload is admitted", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 1)
+				})
+
+				ginkgo.By("creating a mid priority workload which can fit", func() {
+					wl2 = testing.MakeWorkload("wl2", ns.Name).
+						Priority(2).
+						PodSets(*testing.MakePodSet("worker", 1).
+							PreferredTopologyRequest(testing.DefaultBlockTopologyLevel).
+							Obj()).
+						Queue(localQueue.Name).Request(corev1.ResourceCPU, "5").Obj()
+					gomega.Expect(k8sClient.Create(ctx, wl2)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the wl2 gets admitted", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1, wl2)
+					util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 2)
+				})
+
+				ginkgo.By("creating a high priority workload which requires preemption", func() {
+					wl3 = testing.MakeWorkload("wl3", ns.Name).
+						Priority(3).
+						PodSets(*testing.MakePodSet("worker", 2).
+							PreferredTopologyRequest(testing.DefaultBlockTopologyLevel).
+							Obj()).
+						Queue(localQueue.Name).Request(corev1.ResourceCPU, "5").Obj()
+					gomega.Expect(k8sClient.Create(ctx, wl3)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the wl3 gets admitted while wl1 is preempted", func() {
+					util.ExpectWorkloadsToBePreempted(ctx, k8sClient, wl1, wl2)
+					util.FinishEvictionForWorkloads(ctx, k8sClient, wl1, wl2)
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl3)
+					util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 1)
+				})
+			})
+		})
+
 		ginkgo.When("Node structure is mutated during test cases", func() {
 			var (
 				nodes []corev1.Node