Repository: kubernetes-sigs/kueue

Issue:
Flaky e2e tests on v1.32.0 Creating a Job With Queueing Should unsuspend a job and set nodeSelectors
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

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


**What happened**:
Test cases:
- LeaderWorkerSet integration when LeaderWorkerSet created should allow to update the PodTemplate in LeaderWorkerSet
- Kueue when Creating a Job With Queueing Should unsuspend a job and set nodeSelectors
- Kueue when Creating a Job With Queueing Should run with prebuilt workload
- Kueue when Creating a Job With Queueing Should partially admit the Job if configured and not fully fits


flaked:
https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/4567/pull-kueue-test-e2e-main-1-32/1899804437208109056

**What you expected to happen**:

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

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

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

Hints from issue comments:
@PBundyra: /kind flake@mbobrovskyi: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/4543/pull-kueue-test-e2e-main-1-30/1900064219445858304@mimowo: /close
Doing reset of e2e-related flakes as agreed in https://github.com/kubernetes-sigs/kueue/issues/4674#issuecomment-2734095182.

The reason is that we recently bumped up the job resources, and it is expected to help for most of the flakes were attributed to long termination of a job. So, this way we can avoid people looking into an already solved problem.

For more details check the PR [kubernetes/test-infra#34529](https://github.com/kubernetes/test-infra/pull/34529) as discussed here: [#4669](https://github.com/kubernetes-sigs/kueue/issues/4669).

If the failure re-occurs feel free to re-open or open a new one.

Also, feel free to re-open if you have some evidence / hints that constrained resources is not the reason for the failure.@k8s-ci-robot: @mimowo: Closing this issue.

<details>

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

>/close
>Doing reset of e2e-related flakes as agreed in https://github.com/kubernetes-sigs/kueue/issues/4674#issuecomment-2734095182.
>
>The reason is that we recently bumped up the job resources, and it is expected to help for most of the flakes were attributed to long termination of a job. So, this way we can avoid people looking into an already solved problem.
>
>For more details check the PR [kubernetes/test-infra#34529](https://github.com/kubernetes/test-infra/pull/34529) as discussed here: [#4669](https://github.com/kubernetes-sigs/kueue/issues/4669).
>
>If the failure re-occurs feel free to re-open or open a new one.
>
>Also, feel free to re-open if you have some evidence / hints that constrained resources is not the reason for the failure.


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/apis/kueue/v1beta1/localqueue_types.go b/apis/kueue/v1beta1/localqueue_types.go
index 80e48328a..724080fcc 100644
--- a/apis/kueue/v1beta1/localqueue_types.go
+++ b/apis/kueue/v1beta1/localqueue_types.go
@@ -73,6 +73,31 @@ type LocalQueueFlavorStatus struct {
 	// +kubebuilder:validation:MaxItems=8
 	// +optional
 	NodeTaints []corev1.Taint `json:"nodeTaints,omitempty"`
+
+	// topology is the topology that associated with this ResourceFlavor.
+	//
+	// This is an alpha field and requires enabling the TopologyAwareScheduling
+	// feature gate.
+	//
+	// +optional
+	Topology *Topology `json:"topology,omitempty"`
+}
+
+type Topology struct {
+	// name is the name of the topology.
+	//
+	// +required
+	// +kubebuilder:validation:Required
+	Name TopologyReference `json:"name"`
+
+	// levels define the levels of topology.
+	//
+	// +required
+	// +listType=atomic
+	// +kubebuilder:validation:Required
+	// +kubebuilder:validation:MinItems=1
+	// +kubebuilder:validation:MaxItems=8
+	Levels []string `json:"levels"`
 }
 
 // LocalQueueStatus defines the observed state of LocalQueue
diff --git a/apis/kueue/v1beta1/zz_generated.deepcopy.go b/apis/kueue/v1beta1/zz_generated.deepcopy.go
index 829db04d8..19af8c5e5 100644
--- a/apis/kueue/v1beta1/zz_generated.deepcopy.go
+++ b/apis/kueue/v1beta1/zz_generated.deepcopy.go
@@ -630,6 +630,11 @@ func (in *LocalQueueFlavorStatus) DeepCopyInto(out *LocalQueueFlavorStatus) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.Topology != nil {
+		in, out := &in.Topology, &out.Topology
+		*out = new(Topology)
+		(*in).DeepCopyInto(*out)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new LocalQueueFlavorStatus.
@@ -1448,6 +1453,26 @@ func (in *ResourceUsage) DeepCopy() *ResourceUsage {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Topology) DeepCopyInto(out *Topology) {
+	*out = *in
+	if in.Levels != nil {
+		in, out := &in.Levels, &out.Levels
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Topology.
+func (in *Topology) DeepCopy() *Topology {
+	if in == nil {
+		return nil
+	}
+	out := new(Topology)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *TopologyAssignment) DeepCopyInto(out *TopologyAssignment) {
 	*out = *in
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_localqueues.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_localqueues.yaml
index 6da6a186f..fc5bbe7cf 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_localqueues.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_localqueues.yaml
@@ -275,6 +275,30 @@ spec:
                       maxItems: 16
                       type: array
                       x-kubernetes-list-type: set
+                    topology:
+                      description: |-
+                        topology is the topology that associated with this ResourceFlavor.
+
+                        This is an alpha field and requires enabling the TopologyAwareScheduling
+                        feature gate.
+                      properties:
+                        levels:
+                          description: levels define the levels of topology.
+                          items:
+                            type: string
+                          maxItems: 8
+                          minItems: 1
+                          type: array
+                          x-kubernetes-list-type: atomic
+                        name:
+                          description: name is the name of the topology.
+                          maxLength: 253
+                          pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+                          type: string
+                      required:
+                      - levels
+                      - name
+                      type: object
                   required:
                   - name
                   type: object
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml
index 228a57a3c..b0579cd8f 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml
@@ -94,3 +94,26 @@ spec:
         type: object
     served: true
     storage: true
+  - name: v1beta1
+    schema:
+      openAPIV3Schema:
+        properties:
+          levels:
+            description: levels define the levels of topology.
+            items:
+              type: string
+            maxItems: 8
+            minItems: 1
+            type: array
+            x-kubernetes-list-type: atomic
+          name:
+            description: name is the name of the topology.
+            maxLength: 253
+            pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+            type: string
+        required:
+        - name
+        - levels
+        type: object
+    served: true
+    storage: false
diff --git a/client-go/applyconfiguration/kueue/v1beta1/localqueueflavorstatus.go b/client-go/applyconfiguration/kueue/v1beta1/localqueueflavorstatus.go
index 03d1ab24c..622ea7bb1 100644
--- a/client-go/applyconfiguration/kueue/v1beta1/localqueueflavorstatus.go
+++ b/client-go/applyconfiguration/kueue/v1beta1/localqueueflavorstatus.go
@@ -30,6 +30,7 @@ type LocalQueueFlavorStatusApplyConfiguration struct {
 	Resources  []v1.ResourceName                     `json:"resources,omitempty"`
 	NodeLabels map[string]string                     `json:"nodeLabels,omitempty"`
 	NodeTaints []corev1.TaintApplyConfiguration      `json:"nodeTaints,omitempty"`
+	Topology   *TopologyApplyConfiguration           `json:"topology,omitempty"`
 }
 
 // LocalQueueFlavorStatusApplyConfiguration constructs a declarative configuration of the LocalQueueFlavorStatus type for use with
@@ -82,3 +83,11 @@ func (b *LocalQueueFlavorStatusApplyConfiguration) WithNodeTaints(values ...*cor
 	}
 	return b
 }
+
+// WithTopology sets the Topology field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the Topology field is set to the value of the last call.
+func (b *LocalQueueFlavorStatusApplyConfiguration) WithTopology(value *TopologyApplyConfiguration) *LocalQueueFlavorStatusApplyConfiguration {
+	b.Topology = value
+	return b
+}
diff --git a/client-go/applyconfiguration/kueue/v1beta1/topology.go b/client-go/applyconfiguration/kueue/v1beta1/topology.go
new file mode 100644
index 000000000..69289eb1d
--- /dev/null
+++ b/client-go/applyconfiguration/kueue/v1beta1/topology.go
@@ -0,0 +1,53 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+// Code generated by applyconfiguration-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	kueuev1beta1 "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+)
+
+// TopologyApplyConfiguration represents a declarative configuration of the Topology type for use
+// with apply.
+type TopologyApplyConfiguration struct {
+	Name   *kueuev1beta1.TopologyReference `json:"name,omitempty"`
+	Levels []string                        `json:"levels,omitempty"`
+}
+
+// TopologyApplyConfiguration constructs a declarative configuration of the Topology type for use with
+// apply.
+func Topology() *TopologyApplyConfiguration {
+	return &TopologyApplyConfiguration{}
+}
+
+// WithName sets the Name field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the Name field is set to the value of the last call.
+func (b *TopologyApplyConfiguration) WithName(value kueuev1beta1.TopologyReference) *TopologyApplyConfiguration {
+	b.Name = &value
+	return b
+}
+
+// WithLevels adds the given value to the Levels field in the declarative configuration
+// and returns the receiver, so that objects can be build by chaining "With" function invocations.
+// If called multiple times, values provided by each call will be appended to the Levels field.
+func (b *TopologyApplyConfiguration) WithLevels(values ...string) *TopologyApplyConfiguration {
+	for i := range values {
+		b.Levels = append(b.Levels, values[i])
+	}
+	return b
+}
diff --git a/client-go/applyconfiguration/utils.go b/client-go/applyconfiguration/utils.go
index d0b1ffeb0..2cdcea25e 100644
--- a/client-go/applyconfiguration/utils.go
+++ b/client-go/applyconfiguration/utils.go
@@ -137,6 +137,8 @@ func ForKind(kind schema.GroupVersionKind) interface{} {
 		return &kueuev1beta1.ResourceQuotaApplyConfiguration{}
 	case v1beta1.SchemeGroupVersion.WithKind("ResourceUsage"):
 		return &kueuev1beta1.ResourceUsageApplyConfiguration{}
+	case v1beta1.SchemeGroupVersion.WithKind("Topology"):
+		return &kueuev1beta1.TopologyApplyConfiguration{}
 	case v1beta1.SchemeGroupVersion.WithKind("TopologyAssignment"):
 		return &kueuev1beta1.TopologyAssignmentApplyConfiguration{}
 	case v1beta1.SchemeGroupVersion.WithKind("TopologyDomainAssignment"):
diff --git a/config/components/crd/bases/kueue.x-k8s.io_localqueues.yaml b/config/components/crd/bases/kueue.x-k8s.io_localqueues.yaml
index 404f609d3..f502155ee 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_localqueues.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_localqueues.yaml
@@ -260,6 +260,30 @@ spec:
                       maxItems: 16
                       type: array
                       x-kubernetes-list-type: set
+                    topology:
+                      description: |-
+                        topology is the topology that associated with this ResourceFlavor.
+
+                        This is an alpha field and requires enabling the TopologyAwareScheduling
+                        feature gate.
+                      properties:
+                        levels:
+                          description: levels define the levels of topology.
+                          items:
+                            type: string
+                          maxItems: 8
+                          minItems: 1
+                          type: array
+                          x-kubernetes-list-type: atomic
+                        name:
+                          description: name is the name of the topology.
+                          maxLength: 253
+                          pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+                          type: string
+                      required:
+                      - levels
+                      - name
+                      type: object
                   required:
                   - name
                   type: object
diff --git a/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml b/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml
index a543de1d0..a52a41d8d 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml
@@ -79,3 +79,26 @@ spec:
         type: object
     served: true
     storage: true
+  - name: v1beta1
+    schema:
+      openAPIV3Schema:
+        properties:
+          levels:
+            description: levels define the levels of topology.
+            items:
+              type: string
+            maxItems: 8
+            minItems: 1
+            type: array
+            x-kubernetes-list-type: atomic
+          name:
+            description: name is the name of the topology.
+            maxLength: 253
+            pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+            type: string
+        required:
+        - name
+        - levels
+        type: object
+    served: true
+    storage: false
diff --git a/go.mod b/go.mod
index ea169bbe7..6be1ff847 100644
--- a/go.mod
+++ b/go.mod
@@ -12,7 +12,7 @@ require (
 	github.com/onsi/ginkgo/v2 v2.23.0
 	github.com/onsi/gomega v1.36.2
 	github.com/open-policy-agent/cert-controller v0.12.0
-	github.com/project-codeflare/appwrapper v1.0.6
+	github.com/project-codeflare/appwrapper v1.1.0
 	github.com/prometheus/client_golang v1.21.1
 	github.com/prometheus/client_model v0.6.1
 	github.com/ray-project/kuberay/ray-operator v1.2.2
diff --git a/go.sum b/go.sum
index 282a1898c..29fa5a627 100644
--- a/go.sum
+++ b/go.sum
@@ -167,8 +167,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/project-codeflare/appwrapper v1.0.6 h1:byIL9vOFvTAgkAYYQtOcO7DR05Up6iSaH8Gp8UumWi8=
-github.com/project-codeflare/appwrapper v1.0.6/go.mod h1:lNTOQCv4qu05bngU5C6Bl0whXBjP3490FvY4Biy4q7g=
+github.com/project-codeflare/appwrapper v1.1.0 h1:0YDti8qqsnML/YCofOSQ6az++6dhQKYqWh1UexdWrD4=
+github.com/project-codeflare/appwrapper v1.1.0/go.mod h1:Y9XqesMiT0OsHsicHYvC6/Xoj0EMYgJYUo3Vob6JvKo=
 github.com/prometheus/client_golang v1.21.1 h1:DOvXXTqVzvkIewV/CDPFdejpMCGeMcbGCQ8YOmu+Ibk=
 github.com/prometheus/client_golang v1.21.1/go.mod h1:U9NM32ykUErtVBxdvD3zfi+EuFkkaBvMb09mIfe0Zgg=
 github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
diff --git a/hack/e2e-common.sh b/hack/e2e-common.sh
index a7f12485c..a022fee4d 100644
--- a/hack/e2e-common.sh
+++ b/hack/e2e-common.sh
@@ -22,7 +22,7 @@ export YQ="$ROOT_DIR"/bin/yq
 export KIND_VERSION="${E2E_KIND_VERSION/"kindest/node:v"/}"
 
 if [[ -n ${APPWRAPPER_VERSION:-} ]]; then
-    export APPWRAPPER_MANIFEST=${ROOT_DIR}/dep-crds/appwrapper/config/standalone
+    export APPWRAPPER_MANIFEST=${ROOT_DIR}/dep-crds/appwrapper/config/default
     APPWRAPPER_IMAGE=quay.io/ibm/appwrapper:${APPWRAPPER_VERSION}
 fi
 
diff --git a/keps/2724-topology-aware-scheduling/README.md b/keps/2724-topology-aware-scheduling/README.md
index b237ad8c1..6ea92bb75 100644
--- a/keps/2724-topology-aware-scheduling/README.md
+++ b/keps/2724-topology-aware-scheduling/README.md
@@ -27,6 +27,7 @@
   - [User-facing API](#user-facing-api)
   - [Validation](#validation)
   - [Internal APIs](#internal-apis)
+  - [Implicit defaulting of TAS annotations](#implicit-defaulting-of-tas-annotations)
   - [Computing the assignment](#computing-the-assignment)
   - [Enforcing the assignment](#enforcing-the-assignment)
   - [Test Plan](#test-plan)
@@ -621,6 +622,20 @@ for [Beta](#beta). The initial approach for the design is left in the
 [Support for ReplicatedJobs in JobSet](#support-for-replicatedjobs-in-jobset)
 section.
 
+### Implicit defaulting of TAS annotations
+
+Requiring to set the TAS annotations (see [User facing API](#user-facing-api))
+on every PodSet creates a friction for adoption of TAS, and a point of failure.
+
+In order to reduce friction we assume that TAS is used for scheduling workloads
+targetting CQs with only TAS Resource Flavors (RFs with the `spec.topologyName`
+field specified).
+
+A PodSet scheduled with TAS without the explicit annotation is handled as
+if it had the `podset-preferred-topology` annotation pointing to the lowest
+topology level defined in the Topology referenced by the selected TAS flavor.
+We call it "implicit default" as the annotation isn't persisted.
+
 ### Computing the assignment
 
 The extended PodSet assignment is set on admitting the workload. In order to
diff --git a/keps/2724-topology-aware-scheduling/kep.yaml b/keps/2724-topology-aware-scheduling/kep.yaml
index 26776f1a7..289c51900 100644
--- a/keps/2724-topology-aware-scheduling/kep.yaml
+++ b/keps/2724-topology-aware-scheduling/kep.yaml
@@ -31,7 +31,7 @@ milestone:
 # List the feature gate name and the components for which it must be enabled
 feature-gates:
   - name: TopologyAwareScheduling
-  - name: TASLeastAllocated
+  - name: TASMostFreeCapacity
 disable-supported: true
 
 # The following PRR answers are required at beta release
diff --git a/keps/3122-expose-flavors-in-localqueue-status/README.md b/keps/3122-expose-flavors-in-localqueue-status/README.md
index 3fdef6bb4..6b7f3f8cf 100644
--- a/keps/3122-expose-flavors-in-localqueue-status/README.md
+++ b/keps/3122-expose-flavors-in-localqueue-status/README.md
@@ -102,6 +102,35 @@ type LocalQueueFlavorStatus struct {
   // +kubebuilder:validation:MaxItems=8
   // +optional
   NodeTaints []corev1.Taint `json:"nodeTaints,omitempty"`
+
+  // topology is the topology that associated with this ResourceFlavor.
+  //
+  // This is an alpha field and requires enabling the TopologyAwareScheduling
+  // feature gate.
+  //
+  // +optional
+  Topology *Topology `json:"topology,omitempty"`
+}
+```
+
+Add `Topology` API object:
+
+```go
+type Topology struct {
+  // name is the name of the topology.
+  //
+  // +required
+  // +kubebuilder:validation:Required
+  Name TopologyReference `json:"name"`
+
+  // levels define the levels of topology.
+  //
+  // +required
+  // +listType=atomic
+  // +kubebuilder:validation:Required
+  // +kubebuilder:validation:MinItems=1
+  // +kubebuilder:validation:MaxItems=8
+  Levels []string `json:"levels"`
 }
 ```
 
diff --git a/keps/4377-metrics-tls/README.md b/keps/4377-metrics-tls/README.md
new file mode 100644
index 000000000..90b9ca794
--- /dev/null
+++ b/keps/4377-metrics-tls/README.md
@@ -0,0 +1,392 @@
+# KEP-4377: TLS Support for Metrics
+
+<!--
+This is the title of your KEP. Keep it short, simple, and descriptive. A good
+title can help communicate what the KEP is and should be considered as part of
+any review.
+-->
+
+<!--
+A table of contents is helpful for quickly jumping to sections of a KEP and for
+highlighting any additional information provided beyond the standard KEP
+template.
+
+Ensure the TOC is wrapped with
+  <code>&lt;!-- toc --&rt;&lt;!-- /toc --&rt;</code>
+tags, and then generate with `hack/update-toc.sh`.
+-->
+
+<!-- toc -->
+- [Summary](#summary)
+- [Motivation](#motivation)
+  - [Existing Deployments Options](#existing-deployments-options)
+  - [Future Deployment Options](#future-deployment-options)
+  - [Unsupported Deployment Options](#unsupported-deployment-options)
+  - [Goals](#goals)
+  - [Non-Goals](#non-goals)
+- [Proposal](#proposal)
+  - [User Stories (Optional)](#user-stories-optional)
+    - [Story 1](#story-1)
+  - [Notes](#notes)
+  - [Risks and Mitigations](#risks-and-mitigations)
+- [Design Details](#design-details)
+  - [Configuration API Change](#configuration-api-change)
+  - [Deployment changes](#deployment-changes)
+    - [Kustomize](#kustomize)
+    - [Helm](#helm)
+  - [KubeBuilder Recommendation](#kubebuilder-recommendation)
+  - [Prometheus Service Monitor Changes](#prometheus-service-monitor-changes)
+  - [Deployment Changes](#deployment-changes-1)
+  - [Controller Changes](#controller-changes)
+  - [Test Plan](#test-plan)
+      - [Prerequisite testing updates](#prerequisite-testing-updates)
+    - [Unit Tests](#unit-tests)
+    - [Integration tests](#integration-tests)
+  - [Graduation Criteria](#graduation-criteria)
+- [Implementation History](#implementation-history)
+- [Drawbacks](#drawbacks)
+- [Alternatives](#alternatives)
+  - [Opt in API](#opt-in-api)
+<!-- /toc -->
+
+## Summary
+
+Kueue should allow admins to provide their own certificates for metrics endpoints.
+
+## Motivation
+
+Kueue has a few options for managing certificates.
+Kueue provides the ability to manage certificates for both webhooks and metrics.
+For webhooks Kueue provide either external certificates or we use internal certificate rotation.
+
+For metrics Kueue only allows one to use internal certificates.
+In order to have metrics exposed, one needs to turn off tls verify on the service Monitors.
+For many organizations, this can be flagged as a security vulnerability.
+
+We will elaborate on the existing deployment options.
+
+### Existing Deployments Options
+
+Kueue provides the following deployment options for webhooks and metrics.
+Metrics will be assumed to be deployed but they are also optional.
+
+Option 1:
+
+- Admin deploys Kueue with InternalCertManagement set to true.
+- This will use internal certificates for webhooks.
+- Metrics will use internal certificates.
+- ServiceMonitor will skip tls checks via InsecureSkipVerify: true.
+
+Option 2:
+
+- Admin deploys Kueue and Cert Manager (InternalCertManagement set to false).
+- This will use external certificates for webhooks.
+- Metrics will use internal certificates.
+- ServiceMonitor will skip tls checks via InsecureSkipVerify: true.
+
+### Future Deployment Options
+
+Option 3:
+
+- Admin deploys Kueue and Cert Manager (InternalCertManagement set to false).
+- Kueue will use external certificates for webhooks.
+- Kueue will use external certificates for metrics.
+- ServiceMonitor will verify the TLS certificates from a metrics secret created by external certificate solution.
+
+### Unsupported Deployment Options
+
+Not supported Option:
+
+- Admin deploys Kueue and Cert Manager
+- Sets InternalCertManagement to true
+- Cert manager is used to provision certificates for metrics
+- Internal cert rotation is used for webhook.
+- ServiceMonitor will verify the TLS certificates from a metrics secret.
+
+To block this option we will validate that one cannot fall into this scenario.
+
+### Goals
+
+- Provide a way for metrics to verify tls certificates.
+
+### Non-Goals
+
+- Disable secure metrics serving.
+
+It is important to secure metrics if they are enabled.
+We will not provide a way to disable this.
+
+- Allow for external certificates for metrics but internal certificates for webhooks.
+
+- Adopt Internal Cert Manager solution for metrics and provide self signed certificates.
+
+- Supporting APIService certificates
+
+APIService currently uses Insecure flags but there has been no effort made yet to enhance that for Cert manager. It would be good to have support for this but it is out of scope of this KEP.
+
+## Proposal
+
+We will require that if one is using external certificates and prometheus is enabled,
+one must use certificates for metrics. 
+We will effectively be disabling the use of controller runtime generated certificates if an external certificate solution is used.
+
+We will pass this options to the controller runtime `MetricsServer` option where the validation will be done by
+the controller runtime.
+
+### User Stories (Optional)
+
+<!--
+Detail the things that people will be able to do if this KEP is implemented.
+Include as much detail as possible so that people can understand the "how" of
+the system. The goal here is to make this feel real for users without getting
+bogged down.
+-->
+
+#### Story 1
+
+As a cluster adminstration, I want to use an external certificate service for all certificates. 
+I want to use external certificates for webhooks and metrics.
+I am concerned with turning off tls verification for prometheus and I want to validate the certificates.
+
+### Notes
+
+Kueue at the moment skips all tls checking for prometheus endpoints.
+
+```yaml
+# Prometheus Monitor Service (Metrics)
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: controller-manager-metrics-monitor
+  namespace: system
+spec:
+  endpoints:
+    - path: /metrics
+      port: https
+      scheme: https
+      bearerTokenFile: /var/run/secrets/kubernetes.io/serviceaccount/token
+      tlsConfig:
+        insecureSkipVerify: true
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: controller
+      app.kubernetes.io/name: kueue
+```
+
+### Risks and Mitigations
+
+This is more secure option for deploying Kueue but it would require an external dependency on Cert Manager or some other
+certificate management solution.
+
+## Design Details
+
+### Configuration API Change
+
+We will update the configuration to reflect this new state.
+
+```golang
+type InternalCertManagement struct {
+  // Enable controls the use of internal cert management for the webhook
+  // and metrics endpoints.
+  // When enabled Kueue is using libraries to generate and
+  // self-sign the certificates.
+  // When disabled, you need to provide the certificates for
+  // the webhooks and metrics through a third party certificate
+  // This secret is mounted to the kueue controller manager pod. The mount
+  // path for webhooks is /tmp/k8s-webhook-server/serving-certs, whereas for
+  // metrics endpoint the expected path is `/etc/kueue/metrics-certs`.
+  Enable *bool `json:"enable,omitempty"`
+ . . .
+}
+```
+
+### Deployment changes
+
+#### Kustomize
+
+For kustomize based installation, we will add the following optional patches:
+
+- Certificate for metrics
+- Patch to deployment to mount volumes to the deployment
+- ServiceMonitor patches to enable tls checks
+
+These will be enabled similar to how we enable Cert Manager and Prometheus (uncomment those sections).
+
+#### Helm
+
+We will also provide the ability for our Helm charts to use this feature.
+
+The helm chart needs to provide the following abilities:
+
+- Ability to add volumes/volumeMounts to the deployment.
+- Ability to inject custom TlsConfigs for the ServiceMonitor.
+- Ability to provision a certificate for metrics using Cert Manager.
+
+### KubeBuilder Recommendation
+
+Kubebuilder recommends that one enables some kind of external certificate solution for metrics.
+[Recommended For Production](https://book.kubebuilder.io/reference/metrics#recommended-enabling-certificates-for-production-disabled-by-default)
+
+> The default scaffold in cmd/main.go uses a controller-runtime feature to automatically generate a self-signed certificate to secure the metrics server.
+  While this is convenient for development and testing, it is not recommended for production.
+  Those certificates are used to secure the transport layer (TLS). The token authentication using authn/authz, which is enabled by default serves as the application-level credential. However, for example, when you enable the integration of your metrics with Prometheus, those certificates can be used to secure the communication.
+
+With the deprecation of Kube-RBAC-Proxy from Kueue, we are already requiring secure metrics via the use of the below code:
+
+```golang
+if secureMetrics {
+  ...
+  metricsServerOptions.FilterProvider = filters.WithAuthenticationAndAuthorization
+}
+```
+
+### Prometheus Service Monitor Changes
+
+The example below shows:
+
+```yaml
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: controller-manager-metrics-monitor
+  namespace: system
+spec:
+  endpoints:
+    - path: /metrics
+      port: https
+      scheme: https
+      bearerTokenFile: /var/run/secrets/kubernetes.io/serviceaccount/token
+      tlsConfig:
+        serverName: SERVICE_NAME.SERVICE_NAMESPACE.svc
+        insecureSkipVerify: false
+        ca:
+          secret:
+            name: metrics-server-cert
+            key: ca.crt
+        cert:
+          secret:
+            name: metrics-server-cert
+            key: tls.crt
+        keySecret:
+          name: metrics-server-cert
+          key: tls.key
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: controller
+      app.kubernetes.io/name: kueue
+```
+
+### Deployment Changes
+
+Whenever you add Certificate management, one needs to mount a volume to the deployment that references a secret that is created by
+an external service.
+
+Following the kubebuilder examples, one would add the following example to the deployment:
+
+```yaml
+# Add the volumeMount for the metrics-server certs
+- op: add
+  path: /spec/template/spec/containers/0/volumeMounts/-
+  value:
+    mountPath: /etc/k8s-metrics-server/metrics-certs
+    name: metrics-certs
+    readOnly: true
+
+# Add the metrics-server certs volume configuration
+- op: add
+  path: /spec/template/spec/volumes/-
+  value:
+    name: metrics-certs
+    secret:
+      secretName: metrics-server-cert
+      optional: false
+      items:
+        - key: ca.crt
+          path: ca.crt
+        - key: tls.crt
+          path: tls.crt
+        - key: tls.key
+          path: tls.key
+```
+
+### Controller Changes
+
+To keep configurations minimal, we will require that the certificates be mounted to "/etc/kueue/metrics-certs" 
+and the tls certificate and key are called `tls.crt` and `tls.key`, respectively.
+
+```golang
+	if !cfg.InternalCertManager {
+		metricsCertPath := "/etc/kueue/metrics-certs"
+		setupLog.Info("Initializing metrics certificate watcher using provided certificates",
+			"metrics-cert-path", metricsCertPath)
+
+		var err error
+		metricsCertWatcher, err := certwatcher.New(
+			filepath.Join(metricsCertPath, "tls.crt"),
+			filepath.Join(metricsCertPath, "tls.key"),
+		)
+		if err != nil {
+			setupLog.Error(err, "to initialize metrics certificate watcher", "error", err)
+			os.Exit(1)
+		}
+
+		metricsServerOptions.TLSOpts = append(metricsServerOptions.TLSOpts, func(config *tls.Config) {
+			config.GetCertificate = metricsCertWatcher.GetCertificate
+		})
+	}
+```
+
+### Test Plan
+
+<!--
+**Note:** *Not required until targeted at a release.*
+The goal is to ensure that we don't accept enhancements with inadequate testing.
+
+All code is expected to have adequate tests (eventually with coverage
+expectations). Please adhere to the [Kubernetes testing guidelines][testing-guidelines]
+when drafting this test plan.
+
+[testing-guidelines]: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
+-->
+
+[x] I/we understand the owners of the involved components may require updates to
+existing tests to make this code solid enough prior to committing the changes necessary
+to implement this enhancement.
+
+##### Prerequisite testing updates
+
+<!--
+Based on reviewers feedback describe what additional tests need to be added prior
+implementing this enhancement to ensure the enhancements have also solid foundations.
+-->
+
+#### Unit Tests
+
+This will mostly impact cmd/kueue which does not have unit tests.
+
+#### Integration tests
+
+None.
+
+### Graduation Criteria
+
+I think this can go to stable as is as it is an opt-in feature.
+
+## Implementation History
+
+Draft: Feb 25 2025
+
+## Drawbacks
+
+This puts more support burden on upstream Kueue as we had a new pattern for certificates.
+
+## Alternatives
+
+### Opt in API
+
+We proposed an API for metrics to use certificates. This would support the existing solutions but it causes a more complicated support issue.
+Generally, if one is using an external certificate service, they probably do not want to rely on internal certificates.
+
+Due to this, we decided to drop the support of using controller runtime certificates for those that are using Cert Manager.
+
+We keep the same behavior in the case that internal certificates are used.
diff --git a/keps/4377-metrics-tls/kep.yaml b/keps/4377-metrics-tls/kep.yaml
new file mode 100644
index 000000000..566fd89ca
--- /dev/null
+++ b/keps/4377-metrics-tls/kep.yaml
@@ -0,0 +1,29 @@
+title: Certificate Support for Metrics
+kep-number: 4377
+authors:
+  - "@kannon92"
+status: implementable
+creation-date: 2025-02-25
+reviewers:
+  - "@pbundyra"
+  - "@tenzen-y"
+approvers:
+  - "@mimowo"
+  - "@tenzen-y"
+
+see-also:
+  - "NA"
+replaces:
+  - "NA"
+
+# The target maturity stage in the current dev cycle for this KEP.
+stage: stable
+
+# The most recent milestone for which work toward delivery of this KEP has been
+# done. This can be the current (upcoming) milestone, if it is being actively
+# worked on.
+latest-milestone: "v0.11"
+
+# The milestone at which this feature was, or is targeted to be, at each stage.
+milestone:
+  stage: "v0.11"
diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go
index 1869fe036..883ec3483 100644
--- a/pkg/cache/cache.go
+++ b/pkg/cache/cache.go
@@ -766,6 +766,14 @@ func (c *Cache) LocalQueueUsage(qObj *kueue.LocalQueue) (*LocalQueueUsageStats,
 				if rf, ok := c.resourceFlavors[rgFlavor]; ok {
 					flavor.NodeLabels = rf.Spec.NodeLabels
 					flavor.NodeTaints = rf.Spec.NodeTaints
+					if features.Enabled(features.TopologyAwareScheduling) && rf.Spec.TopologyName != nil {
+						if topology, ok := c.tasCache.flavors[rgFlavor]; ok {
+							flavor.Topology = &kueue.Topology{
+								Name:   topology.TopologyName,
+								Levels: topology.Levels,
+							}
+						}
+					}
 				}
 				flavors = append(flavors, flavor)
 			}
diff --git a/pkg/cache/clusterqueue.go b/pkg/cache/clusterqueue.go
index b79ac2e1f..3149d35d3 100644
--- a/pkg/cache/clusterqueue.go
+++ b/pkg/cache/clusterqueue.go
@@ -508,7 +508,7 @@ func (c *clusterQueue) updateWorkloadUsage(wi *workload.Info, m int64) {
 			removeUsage(c, fr, q)
 		}
 	}
-	if features.Enabled(features.TopologyAwareScheduling) && wi.IsRequestingTAS() {
+	if features.Enabled(features.TopologyAwareScheduling) && wi.IsUsingTAS() {
 		for tasFlavor, tasUsage := range wi.TASUsage() {
 			if tasFlvCache := c.tasFlavorCache(tasFlavor); tasFlvCache != nil {
 				if m == 1 {
@@ -627,3 +627,14 @@ func workloadBelongsToLocalQueue(wl *kueue.Workload, q *kueue.LocalQueue) bool {
 func (c *clusterQueue) fairWeight() *resource.Quantity {
 	return &c.FairWeight
 }
+
+func (c *clusterQueue) isTASOnly() bool {
+	for _, rg := range c.ResourceGroups {
+		for _, fName := range rg.Flavors {
+			if _, found := c.tasFlavors[fName]; !found {
+				return false
+			}
+		}
+	}
+	return true
+}
diff --git a/pkg/cache/clusterqueue_snapshot.go b/pkg/cache/clusterqueue_snapshot.go
index a628cd2ac..16548e2e5 100644
--- a/pkg/cache/clusterqueue_snapshot.go
+++ b/pkg/cache/clusterqueue_snapshot.go
@@ -55,6 +55,7 @@ type ClusterQueueSnapshot struct {
 	hierarchy.ClusterQueue[*CohortSnapshot]
 
 	TASFlavors map[kueue.ResourceFlavorReference]*TASFlavorSnapshot
+	tasOnly    bool
 }
 
 // RGByResource returns the ResourceGroup which contains capacity
@@ -214,3 +215,7 @@ func (c *ClusterQueueSnapshot) FindTopologyAssignmentsForWorkload(
 	}
 	return result
 }
+
+func (c *ClusterQueueSnapshot) IsTASOnly() bool {
+	return c.tasOnly
+}
diff --git a/pkg/cache/snapshot.go b/pkg/cache/snapshot.go
index c74df1bfc..a7507b9dc 100644
--- a/pkg/cache/snapshot.go
+++ b/pkg/cache/snapshot.go
@@ -173,6 +173,7 @@ func snapshotClusterQueue(c *clusterQueue) *ClusterQueueSnapshot {
 		AdmissionChecks:               utilmaps.DeepCopySets[kueue.ResourceFlavorReference](c.AdmissionChecks),
 		ResourceNode:                  c.resourceNode.Clone(),
 		TASFlavors:                    make(map[kueue.ResourceFlavorReference]*TASFlavorSnapshot),
+		tasOnly:                       c.isTASOnly(),
 	}
 	for i, rg := range c.ResourceGroups {
 		cc.ResourceGroups[i] = rg.Clone()
diff --git a/pkg/cache/tas_flavor_snapshot.go b/pkg/cache/tas_flavor_snapshot.go
index 096dbda41..358d4ce91 100644
--- a/pkg/cache/tas_flavor_snapshot.go
+++ b/pkg/cache/tas_flavor_snapshot.go
@@ -27,6 +27,7 @@ import (
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
+	"k8s.io/utils/ptr"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -158,7 +159,11 @@ func (s *TASFlavorSnapshot) addNode(node corev1.Node) utiltas.TopologyDomainID {
 }
 
 func (s *TASFlavorSnapshot) isLowestLevelNode() bool {
-	return s.levelKeys[len(s.levelKeys)-1] == corev1.LabelHostname
+	return s.lowestLevel() == corev1.LabelHostname
+}
+
+func (s *TASFlavorSnapshot) lowestLevel() string {
+	return s.levelKeys[len(s.levelKeys)-1]
 }
 
 // initialize prepares the topology tree structure. This structure holds
@@ -311,6 +316,7 @@ type TASPodSetRequests struct {
 	SinglePodRequests resources.Requests
 	Count             int32
 	Flavor            kueue.ResourceFlavorReference
+	Implied           bool
 }
 
 func (t *TASPodSetRequests) TotalRequests() resources.Requests {
@@ -402,13 +408,12 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	tasPodSetRequests TASPodSetRequests,
 	assumedUsage map[utiltas.TopologyDomainID]resources.Requests,
 	simulateEmpty bool) (*kueue.TopologyAssignment, string) {
-	topologyRequest := tasPodSetRequests.PodSet.TopologyRequest
 	requests := tasPodSetRequests.SinglePodRequests.Clone()
 	requests.Add(resources.Requests{corev1.ResourcePods: 1})
 	podSetTolerations := tasPodSetRequests.PodSet.Template.Spec.Tolerations
 	count := tasPodSetRequests.Count
-	required := topologyRequest.Required != nil
-	key := levelKey(topologyRequest)
+	required := isRequired(tasPodSetRequests.PodSet.TopologyRequest)
+	key := s.levelKeyWithImpliedFallback(&tasPodSetRequests)
 	if key == nil {
 		return nil, "topology level not specified"
 	}
@@ -454,7 +459,24 @@ func (s *TASFlavorSnapshot) resolveLevelIdx(levelKey string) (int, bool) {
 	return levelIdx, true
 }
 
+func isRequired(tr *kueue.PodSetTopologyRequest) bool {
+	return tr != nil && tr.Required != nil
+}
+
+func (s *TASFlavorSnapshot) levelKeyWithImpliedFallback(tasRequests *TASPodSetRequests) *string {
+	if key := levelKey(tasRequests.PodSet.TopologyRequest); key != nil {
+		return key
+	}
+	if tasRequests.Implied {
+		return ptr.To(s.lowestLevel())
+	}
+	return nil
+}
+
 func levelKey(topologyRequest *kueue.PodSetTopologyRequest) *string {
+	if topologyRequest == nil {
+		return nil
+	}
 	if topologyRequest.Required != nil {
 		return topologyRequest.Required
 	} else if topologyRequest.Preferred != nil {
@@ -463,20 +485,20 @@ func levelKey(topologyRequest *kueue.PodSetTopologyRequest) *string {
 	return nil
 }
 
-// findMostAllocatedDomainIdx finds an index of the first domain with the lowest
+// findBestFitDomainIdx finds an index of the first domain with the lowest
 // value of state, higher or equal than count.
 // If such a domain doesn't exist, it returns 0 as it's an index of the domain with the
 // most available resources
-func findMostAllocatedDomainIdx(domains []*domain, count int32) int {
-	mostAllocatedFitIdx := 0
+func findBestFitDomainIdx(domains []*domain, count int32) int {
+	bestFitIdx := 0
 	for i, domain := range domains {
-		if domain.state >= count && domain.state != domains[mostAllocatedFitIdx].state {
+		if domain.state >= count && domain.state != domains[bestFitIdx].state {
 			// choose the first occurrence of fitting domains
 			// to make it consecutive with other podSet's
-			mostAllocatedFitIdx = i
+			bestFitIdx = i
 		}
 	}
-	return mostAllocatedFitIdx
+	return bestFitIdx
 }
 
 func (s *TASFlavorSnapshot) findLevelWithFitDomains(levelIdx int, required bool, count int32) (int, []*domain, string) {
@@ -487,8 +509,8 @@ func (s *TASFlavorSnapshot) findLevelWithFitDomains(levelIdx int, required bool,
 	levelDomains := slices.Collect(maps.Values(domains))
 	sortedDomain := s.sortedDomains(levelDomains)
 	topDomain := sortedDomain[0]
-	if !features.Enabled(features.TASLeastAllocated) && topDomain.state >= count {
-		topDomain = sortedDomain[findMostAllocatedDomainIdx(sortedDomain, count)]
+	if !features.Enabled(features.TASMostFreeCapacity) && topDomain.state >= count {
+		topDomain = sortedDomain[findBestFitDomainIdx(sortedDomain, count)]
 	}
 	if topDomain.state < count {
 		if required {
@@ -501,8 +523,8 @@ func (s *TASFlavorSnapshot) findLevelWithFitDomains(levelIdx int, required bool,
 		remainingCount := count
 		for idx := 0; remainingCount > 0 && idx < len(sortedDomain) && sortedDomain[idx].state > 0; idx++ {
 			offset := 0
-			if !features.Enabled(features.TASLeastAllocated) && sortedDomain[idx].state >= remainingCount {
-				offset = findMostAllocatedDomainIdx(sortedDomain[idx:], remainingCount)
+			if !features.Enabled(features.TASMostFreeCapacity) && sortedDomain[idx].state >= remainingCount {
+				offset = findBestFitDomainIdx(sortedDomain[idx:], remainingCount)
 			}
 			results = append(results, sortedDomain[idx+offset])
 			remainingCount -= sortedDomain[idx].state
@@ -519,9 +541,9 @@ func (s *TASFlavorSnapshot) updateCountsToMinimum(domains []*domain, count int32
 	result := make([]*domain, 0)
 	remainingCount := count
 	for i, domain := range domains {
-		if !features.Enabled(features.TASLeastAllocated) && domain.state >= remainingCount {
-			mostAllocatedIdx := findMostAllocatedDomainIdx(domains[i:], remainingCount)
-			domain = domains[i+mostAllocatedIdx]
+		if !features.Enabled(features.TASMostFreeCapacity) && domain.state >= remainingCount {
+			bestFitIdx := findBestFitDomainIdx(domains[i:], remainingCount)
+			domain = domains[i+bestFitIdx]
 		}
 
 		if domain.state >= remainingCount {
diff --git a/pkg/controller/jobframework/setup.go b/pkg/controller/jobframework/setup.go
index 18981a7f0..1f6c7c03e 100644
--- a/pkg/controller/jobframework/setup.go
+++ b/pkg/controller/jobframework/setup.go
@@ -85,11 +85,17 @@ func (m *integrationManager) setupControllers(ctx context.Context, mgr ctrl.Mana
 				if !meta.IsNoMatchError(err) {
 					return fmt.Errorf("%s: %w", fwkNamePrefix, err)
 				}
-				logger.Info("No matching API in the server for job framework, skipped setup of controller and webhook")
+				// Webhook must be registered now; controller can be registered later.
+				// The issue is that the controller-runtime silently ignores attempts to update the webhook
+				// for an endpoint that already has one and we don't want the NoopWebhook to be installed.
+				if err := cb.SetupWebhook(mgr, opts...); err != nil {
+					return fmt.Errorf("%s: unable to create webhook: %w", fwkNamePrefix, err)
+				}
+				logger.Info("No matching API in the server for job framework, deferring setting up controller")
 				go waitForAPI(ctx, mgr, log, gvk, func() {
-					log.Info("API now available, starting controller and webhook", "gvk", gvk)
+					log.Info("API now available, starting controller", "gvk", gvk)
 					if err := m.setupControllerAndWebhook(mgr, name, fwkNamePrefix, cb, options, opts...); err != nil {
-						log.Error(err, "Failed to setup controller and webhook for job framework")
+						log.Error(err, "Failed to setup controller for job framework")
 					}
 				})
 			} else {
diff --git a/pkg/controller/jobframework/validation.go b/pkg/controller/jobframework/validation.go
index aee71b0c4..fe2d046e9 100644
--- a/pkg/controller/jobframework/validation.go
+++ b/pkg/controller/jobframework/validation.go
@@ -57,7 +57,7 @@ var (
 		rayv1.SchemeGroupVersion.WithKind("RayJob").String(),
 		corev1.SchemeGroupVersion.WithKind("Pod").String(),
 		rayv1.SchemeGroupVersion.WithKind("RayCluster").String(),
-		awv1beta2.GroupVersion.WithKind("AppWrapper").String(),
+		awv1beta2.GroupVersion.WithKind(awv1beta2.AppWrapperKind).String(),
 	)
 )
 
diff --git a/pkg/controller/jobs/appwrapper/appwrapper_controller.go b/pkg/controller/jobs/appwrapper/appwrapper_controller.go
index 76ca00bb7..39b6f7c3a 100644
--- a/pkg/controller/jobs/appwrapper/appwrapper_controller.go
+++ b/pkg/controller/jobs/appwrapper/appwrapper_controller.go
@@ -41,7 +41,7 @@ import (
 )
 
 var (
-	gvk = awv1beta2.GroupVersion.WithKind("AppWrapper")
+	gvk = awv1beta2.GroupVersion.WithKind(awv1beta2.AppWrapperKind)
 
 	FrameworkName = "workload.codeflare.dev/appwrapper"
 
@@ -86,7 +86,7 @@ func NewJob() jobframework.GenericJob {
 }
 
 func isAppWrapper(owner *metav1.OwnerReference) bool {
-	return owner.Kind == "AppWrapper" && strings.HasPrefix(owner.APIVersion, gvk.Group)
+	return owner.Kind == awv1beta2.AppWrapperKind && strings.HasPrefix(owner.APIVersion, gvk.Group)
 }
 
 func SetupIndexes(ctx context.Context, indexer client.FieldIndexer) error {
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index 7d2171608..4227e94c2 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -150,7 +150,7 @@ const (
 	// kep: https://github.com/kubernetes-sigs/kueue/tree/main/keps/2724-topology-aware-scheduling
 	//
 	// Enable to set use LeastAlloactedFit algorithm for TAS
-	TASLeastAllocated featuregate.Feature = "TASLeastAllocated"
+	TASMostFreeCapacity featuregate.Feature = "TASMostFreeCapacity"
 )
 
 func init() {
@@ -231,7 +231,7 @@ var defaultVersionedFeatureGates = map[featuregate.Feature]featuregate.Versioned
 	LocalQueueDefaulting: {
 		{Version: version.MustParse("0.10"), Default: false, PreRelease: featuregate.Alpha},
 	},
-	TASLeastAllocated: {
+	TASMostFreeCapacity: {
 		{Version: version.MustParse("0.11"), Default: false, PreRelease: featuregate.Deprecated},
 	},
 }
diff --git a/pkg/scheduler/flavorassigner/tas_flavorassigner.go b/pkg/scheduler/flavorassigner/tas_flavorassigner.go
index 45f65be36..93f6f8159 100644
--- a/pkg/scheduler/flavorassigner/tas_flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/tas_flavorassigner.go
@@ -31,13 +31,14 @@ import (
 func (a *Assignment) WorkloadsTopologyRequests(wl *workload.Info, cq *cache.ClusterQueueSnapshot) cache.WorkloadTASRequests {
 	tasRequests := make(cache.WorkloadTASRequests)
 	for i, podSet := range wl.Obj.Spec.PodSets {
-		if podSet.TopologyRequest != nil {
+		if isTASRequested(&podSet, cq) {
 			psAssignment := a.podSetAssignmentByName(podSet.Name)
 			if psAssignment.Status.IsError() {
 				// There is no resource quota assignment for the PodSet - no need to check TAS.
 				continue
 			}
-			psTASRequest, err := podSetTopologyRequest(psAssignment, wl, cq, i)
+			isTASImplied := isTASImplied(&podSet, cq)
+			psTASRequest, err := podSetTopologyRequest(psAssignment, wl, cq, isTASImplied, i)
 			if err != nil {
 				psAssignment.error(err)
 			} else {
@@ -51,6 +52,7 @@ func (a *Assignment) WorkloadsTopologyRequests(wl *workload.Info, cq *cache.Clus
 func podSetTopologyRequest(psAssignment *PodSetAssignment,
 	wl *workload.Info,
 	cq *cache.ClusterQueueSnapshot,
+	isTASImplied bool,
 	podSetIndex int) (*cache.TASPodSetRequests, error) {
 	if len(cq.TASFlavors) == 0 {
 		return nil, errors.New("workload requires Topology, but there is no TAS cache information")
@@ -71,6 +73,7 @@ func podSetTopologyRequest(psAssignment *PodSetAssignment,
 		SinglePodRequests: singlePodRequests,
 		PodSet:            podSet,
 		Flavor:            *tasFlvr,
+		Implied:           isTASImplied,
 	}, nil
 }
 
@@ -107,9 +110,25 @@ func checkPodSetAndFlavorMatchForTAS(cq *cache.ClusterQueueSnapshot, ps *kueue.P
 			return ptr.To(fmt.Sprintf("Flavor %q does not contain the requested level", flavor.Name))
 		}
 	}
+	// If this is a TAS-only CQ, then no TopologyRequest is ok
+	if isTASImplied(ps, cq) {
+		return nil
+	}
 	// For PodSets which don't use TAS skip resource flavors which are only for TAS
 	if ps.TopologyRequest == nil && flavor.Spec.TopologyName != nil {
 		return ptr.To(fmt.Sprintf("Flavor %q supports only TopologyAwareScheduling", flavor.Name))
 	}
 	return nil
 }
+
+// isTASImplied returns true if TAS is requested implicitly - there is no
+// explicit
+func isTASImplied(ps *kueue.PodSet, cq *cache.ClusterQueueSnapshot) bool {
+	return ps.TopologyRequest == nil && cq.IsTASOnly()
+}
+
+// isTASRequested checks if TAS is requested for the input PodSet, either
+// explicitly or implicitly.
+func isTASRequested(ps *kueue.PodSet, cq *cache.ClusterQueueSnapshot) bool {
+	return ps.TopologyRequest != nil || cq.IsTASOnly()
+}
diff --git a/pkg/scheduler/preempted_workloads.go b/pkg/scheduler/preemption/preempted_workloads.go
similarity index 76%
rename from pkg/scheduler/preempted_workloads.go
rename to pkg/scheduler/preemption/preempted_workloads.go
index 7476f4e2f..ad532d82f 100644
--- a/pkg/scheduler/preempted_workloads.go
+++ b/pkg/scheduler/preemption/preempted_workloads.go
@@ -14,16 +14,15 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package scheduler
+package preemption
 
 import (
-	"sigs.k8s.io/kueue/pkg/scheduler/preemption"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
-type preemptedWorkloads map[string]*workload.Info
+type PreemptedWorkloads map[string]*workload.Info
 
-func (p preemptedWorkloads) hasAny(newTargets []*preemption.Target) bool {
+func (p PreemptedWorkloads) HasAny(newTargets []*Target) bool {
 	for _, target := range newTargets {
 		if _, found := p[workload.Key(target.WorkloadInfo.Obj)]; found {
 			return true
@@ -32,7 +31,7 @@ func (p preemptedWorkloads) hasAny(newTargets []*preemption.Target) bool {
 	return false
 }
 
-func (p preemptedWorkloads) insert(newTargets []*preemption.Target) {
+func (p PreemptedWorkloads) Insert(newTargets []*Target) {
 	for _, target := range newTargets {
 		p[workload.Key(target.WorkloadInfo.Obj)] = target.WorkloadInfo
 	}
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 0a2d41123..84d51cdb6 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -159,7 +159,7 @@ func (p *Preemptor) getTargets(preemptionCtx *preemptionCtx) []*Target {
 	}
 
 	if p.enableFairSharing {
-		return p.fairPreemptions(preemptionCtx, candidates)
+		return fairPreemptions(preemptionCtx, candidates, p.fsStrategies)
 	}
 	// There is a potential of preemption of workloads from the other queue in the
 	// cohort. We proceed with borrowing only if the dedicated policy
@@ -375,17 +375,16 @@ func parseStrategies(s []config.PreemptionStrategy) []fsStrategy {
 	return strategies
 }
 
-func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info) []*Target {
-	if logV := preemptionCtx.log.V(5); logV.Enabled() {
-		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
-	}
+// runFirstFsStrategy runs the first configured FairSharing strategy,
+// and returns (fits, targets, retryCandidates) retryCandidates may be
+// used if rule S2-b is configured.
+func runFirstFsStrategy(preemptionCtx *preemptionCtx, candidates []*workload.Info, strategy fsStrategy) (bool, []*Target, []*workload.Info) {
 	requests := preemptionCtx.requests
 	cqHeap := cqHeapFromCandidates(candidates, false, preemptionCtx.snapshot)
 	newNominatedShareValue := preemptionCtx.preemptorCQ.DominantResourceShareWith(requests)
 	var targets []*Target
-	fits := false
 	var retryCandidates []*workload.Info
-	for cqHeap.Len() > 0 && !fits {
+	for cqHeap.Len() > 0 {
 		candCQ := cqHeap.Pop()
 
 		if candCQ.cq == preemptionCtx.preemptorCQ {
@@ -396,8 +395,7 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*
 				Reason:       kueue.InClusterQueueReason,
 			})
 			if workloadFits(preemptionCtx, true) {
-				fits = true
-				break
+				return true, targets, nil
 			}
 			newNominatedShareValue = preemptionCtx.preemptorCQ.DominantResourceShareWith(requests)
 			candCQ.workloads = candCQ.workloads[1:]
@@ -410,8 +408,7 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*
 
 		for i, candWl := range candCQ.workloads {
 			newCandShareVal := candCQ.cq.DominantResourceShareWithout(candWl.FlavorResourceUsage())
-			strategy := p.fsStrategies[0](newNominatedShareValue, candCQ.share, newCandShareVal)
-			if strategy {
+			if strategy(newNominatedShareValue, candCQ.share, newCandShareVal) {
 				preemptionCtx.snapshot.RemoveWorkload(candWl)
 				reason := kueue.InCohortFairSharingReason
 
@@ -420,8 +417,7 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*
 					Reason:       reason,
 				})
 				if workloadFits(preemptionCtx, true) {
-					fits = true
-					break
+					return true, targets, nil
 				}
 				candCQ.workloads = candCQ.workloads[i+1:]
 				if len(candCQ.workloads) > 0 && cqIsBorrowing(candCQ.cq, preemptionCtx.frsNeedPreemption) {
@@ -435,30 +431,46 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*
 			}
 		}
 	}
-	if !fits && len(p.fsStrategies) > 1 {
-		// Try next strategy if the previous strategy wasn't enough
-		cqHeap = cqHeapFromCandidates(retryCandidates, true, preemptionCtx.snapshot)
-
-		for cqHeap.Len() > 0 && !fits {
-			candCQ := cqHeap.Pop()
-			// Due to API validation, we can only reach here if the second strategy is LessThanInitialShare,
-			// in which case the last parameter for the strategy function is irrelevant.
-			if p.fsStrategies[1](newNominatedShareValue, candCQ.share, 0) {
-				// The criteria doesn't depend on the preempted workload, so just preempt the first candidate.
-				candWl := candCQ.workloads[0]
-				preemptionCtx.snapshot.RemoveWorkload(candWl)
-				targets = append(targets, &Target{
-					WorkloadInfo: candWl,
-					Reason:       kueue.InCohortFairSharingReason,
-				})
-				if workloadFits(preemptionCtx, true) {
-					fits = true
-				}
-				// No requeueing because there doesn't seem to be an scenario where
-				// it's possible to apply rule S2-b more than once in a CQ.
+	return false, targets, retryCandidates
+}
+
+// runSecondFsStrategy implements Fair Sharing Rule S2-b. It returns
+// (fits, targets).
+func runSecondFsStrategy(retryCandidates []*workload.Info, preemptionCtx *preemptionCtx, targets []*Target) (bool, []*Target) {
+	cqHeap := cqHeapFromCandidates(retryCandidates, true, preemptionCtx.snapshot)
+	newNominatedShareValue := preemptionCtx.preemptorCQ.DominantResourceShareWith(preemptionCtx.requests)
+	for cqHeap.Len() > 0 {
+		candCQ := cqHeap.Pop()
+		// Due to API validation, we can only reach here if the second strategy is LessThanInitialShare,
+		// in which case the last parameter for the strategy function is irrelevant.
+		if lessThanInitialShare(newNominatedShareValue, candCQ.share, 0) {
+			// The criteria doesn't depend on the preempted workload, so just preempt the first candidate.
+			candWl := candCQ.workloads[0]
+			preemptionCtx.snapshot.RemoveWorkload(candWl)
+			targets = append(targets, &Target{
+				WorkloadInfo: candWl,
+				Reason:       kueue.InCohortFairSharingReason,
+			})
+			if workloadFits(preemptionCtx, true) {
+				return true, targets
 			}
+			// No requeueing because there doesn't seem to be an scenario where
+			// it's possible to apply rule S2-b more than once in a CQ.
 		}
 	}
+	return false, targets
+}
+
+func fairPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info, strategies []fsStrategy) []*Target {
+	if logV := preemptionCtx.log.V(5); logV.Enabled() {
+		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
+	}
+
+	fits, targets, retryCandidates := runFirstFsStrategy(preemptionCtx, candidates, strategies[0])
+	if !fits && len(strategies) > 1 {
+		fits, targets = runSecondFsStrategy(retryCandidates, preemptionCtx, targets)
+	}
+
 	if !fits {
 		restoreSnapshot(preemptionCtx.snapshot, targets)
 		return nil
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 1ce869597..906f634fc 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -206,7 +206,7 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 	// This is because there can be other workloads deeper in a clusterQueue whose
 	// head got admitted that should be scheduled in the cohort before the heads
 	// of other clusterQueues.
-	preemptedWorkloads := make(preemptedWorkloads)
+	preemptedWorkloads := make(preemption.PreemptedWorkloads)
 	skippedPreemptions := make(map[kueue.ClusterQueueReference]int)
 	for iterator.hasNext() {
 		e := iterator.pop()
@@ -235,7 +235,7 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 		}
 
 		// We skip multiple-preemptions per cohort if any of the targets are overlapping
-		if preemptedWorkloads.hasAny(e.preemptionTargets) {
+		if preemptedWorkloads.HasAny(e.preemptionTargets) {
 			setSkipped(e, "Workload has overlapping preemption targets with another workload")
 			skippedPreemptions[cq.Name]++
 			continue
@@ -249,7 +249,7 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 			}
 			continue
 		}
-		preemptedWorkloads.insert(e.preemptionTargets)
+		preemptedWorkloads.Insert(e.preemptionTargets)
 		cq.AddUsage(usage)
 
 		if e.assignment.RepresentativeMode() == flavorassigner.Preempt {
@@ -369,7 +369,7 @@ func (s *Scheduler) nominate(ctx context.Context, workloads []workload.Info, sna
 	return entries
 }
 
-func fits(cq *cache.ClusterQueueSnapshot, usage *workload.Usage, preemptedWorkloads preemptedWorkloads, newTargets []*preemption.Target) bool {
+func fits(cq *cache.ClusterQueueSnapshot, usage *workload.Usage, preemptedWorkloads preemption.PreemptedWorkloads, newTargets []*preemption.Target) bool {
 	workloads := slices.Collect(maps.Values(preemptedWorkloads))
 	for _, target := range newTargets {
 		workloads = append(workloads, target.WorkloadInfo)
@@ -460,7 +460,7 @@ func (s *Scheduler) getInitialAssignments(log logr.Logger, wl *workload.Info, sn
 }
 
 func updateAssignmentForTAS(cq *cache.ClusterQueueSnapshot, wl *workload.Info, assignment *flavorassigner.Assignment, targets []*preemption.Target) {
-	if features.Enabled(features.TopologyAwareScheduling) && assignment.RepresentativeMode() == flavorassigner.Preempt && wl.IsRequestingTAS() {
+	if features.Enabled(features.TopologyAwareScheduling) && assignment.RepresentativeMode() == flavorassigner.Preempt && (wl.IsRequestingTAS() || cq.IsTASOnly()) {
 		tasRequests := assignment.WorkloadsTopologyRequests(wl, cq)
 		var tasResult cache.TASAssignmentsResult
 		if len(targets) > 0 {
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 4443b75a9..44ed6cd62 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -281,6 +281,14 @@ func dropExcludedResources(input corev1.ResourceList, excludedPrefixes []string)
 	return res
 }
 
+// IsUsingTAS returns information if the workload is using TAS
+func (i *Info) IsUsingTAS() bool {
+	return slices.ContainsFunc(i.TotalRequests,
+		func(ps PodSetResources) bool {
+			return ps.TopologyRequest != nil
+		})
+}
+
 // IsRequestingTAS returns information if the workload is requesting TAS
 func (i *Info) IsRequestingTAS() bool {
 	return slices.ContainsFunc(i.Obj.Spec.PodSets,
@@ -291,7 +299,7 @@ func (i *Info) IsRequestingTAS() bool {
 
 // TASUsage returns topology usage requested by the Workload
 func (i *Info) TASUsage() TASUsage {
-	if !features.Enabled(features.TopologyAwareScheduling) || !i.IsRequestingTAS() {
+	if !features.Enabled(features.TopologyAwareScheduling) || !i.IsUsingTAS() {
 		return nil
 	}
 	result := make(TASUsage, 0)
diff --git a/site/content/en/docs/reference/kueue.v1beta1.md b/site/content/en/docs/reference/kueue.v1beta1.md
index 0edccabd4..b922e3876 100644
--- a/site/content/en/docs/reference/kueue.v1beta1.md
+++ b/site/content/en/docs/reference/kueue.v1beta1.md
@@ -1288,6 +1288,15 @@ have the same labels.</p>
 have.</p>
 </td>
 </tr>
+<tr><td><code>topology</code><br/>
+<a href="#kueue-x-k8s-io-v1beta1-Topology"><code>Topology</code></a>
+</td>
+<td>
+   <p>topology is the topology that associated with this ResourceFlavor.</p>
+<p>This is an alpha field and requires enabling the TopologyAwareScheduling
+feature gate.</p>
+</td>
+</tr>
 </tbody>
 </table>
 
@@ -2367,6 +2376,37 @@ words, it's the used quota that is over the nominalQuota.</p>
 
 
 
+## `Topology`     {#kueue-x-k8s-io-v1beta1-Topology}
+    
+
+**Appears in:**
+
+- [LocalQueueFlavorStatus](#kueue-x-k8s-io-v1beta1-LocalQueueFlavorStatus)
+
+
+
+<table class="table">
+<thead><tr><th width="30%">Field</th><th>Description</th></tr></thead>
+<tbody>
+    
+  
+<tr><td><code>name</code> <B>[Required]</B><br/>
+<a href="#kueue-x-k8s-io-v1beta1-TopologyReference"><code>TopologyReference</code></a>
+</td>
+<td>
+   <p>name is the name of the topology.</p>
+</td>
+</tr>
+<tr><td><code>levels</code> <B>[Required]</B><br/>
+<code>[]string</code>
+</td>
+<td>
+   <p>levels define the levels of topology.</p>
+</td>
+</tr>
+</tbody>
+</table>
+
 ## `TopologyAssignment`     {#kueue-x-k8s-io-v1beta1-TopologyAssignment}
     
 
@@ -2443,6 +2483,8 @@ domain indicated by the values field.</p>
 
 - [ResourceFlavorSpec](#kueue-x-k8s-io-v1beta1-ResourceFlavorSpec)
 
+- [Topology](#kueue-x-k8s-io-v1beta1-Topology)
+
 
 <p>TopologyReference is the name of the Topology.</p>
 
diff --git a/vendor/github.com/project-codeflare/appwrapper/api/v1beta2/groupversion_info.go b/vendor/github.com/project-codeflare/appwrapper/api/v1beta2/groupversion_info.go
index cd8fa9565..c17210596 100644
--- a/vendor/github.com/project-codeflare/appwrapper/api/v1beta2/groupversion_info.go
+++ b/vendor/github.com/project-codeflare/appwrapper/api/v1beta2/groupversion_info.go
@@ -28,6 +28,9 @@ var (
 	// GroupVersion is group version used to register these objects
 	GroupVersion = schema.GroupVersion{Group: "workload.codeflare.dev", Version: "v1beta2"}
 
+	// AppWrapperKind is the kind name
+	AppWrapperKind = "AppWrapper"
+
 	// SchemeBuilder is used to add go types to the GroupVersionKind scheme
 	SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion}
 
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 954d9a1dd..267956b65 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -283,7 +283,7 @@ github.com/peterbourgon/diskv
 # github.com/pkg/errors v0.9.1
 ## explicit
 github.com/pkg/errors
-# github.com/project-codeflare/appwrapper v1.0.6
+# github.com/project-codeflare/appwrapper v1.1.0
 ## explicit; go 1.23.0
 github.com/project-codeflare/appwrapper/api/v1beta2
 github.com/project-codeflare/appwrapper/pkg/utils



Test patch:
diff --git a/pkg/cache/tas_cache_test.go b/pkg/cache/tas_cache_test.go
index 85891336c..3918717e6 100644
--- a/pkg/cache/tas_cache_test.go
+++ b/pkg/cache/tas_cache_test.go
@@ -225,21 +225,21 @@ func TestFindTopologyAssignment(t *testing.T) {
 	}
 
 	cases := map[string]struct {
-		// TODO: remove after dropping the TASLeastAllocated feature gate
-		enableTASLeastAllocated bool
-		wantReason              string
-		topologyRequest         kueue.PodSetTopologyRequest
-		levels                  []string
-		nodeLabels              map[string]string
-		nodes                   []corev1.Node
-		pods                    []corev1.Pod
-		requests                resources.Requests
-		count                   int32
-		tolerations             []corev1.Toleration
-		wantAssignment          *kueue.TopologyAssignment
+		// TODO: remove after dropping the TASMostFreeCapacity feature gate
+		enableTASMostFreeCapacity bool
+		wantReason                string
+		topologyRequest           kueue.PodSetTopologyRequest
+		levels                    []string
+		nodeLabels                map[string]string
+		nodes                     []corev1.Node
+		pods                      []corev1.Pod
+		requests                  resources.Requests
+		count                     int32
+		tolerations               []corev1.Toleration
+		wantAssignment            *kueue.TopologyAssignment
 	}{
-		// TODO: remove suffixes LeastAllocated/MostAllocated after dropping the TASLeastAllocated feature gate
-		"minimize the number of used racks before optimizing the number of nodes; LeastAllocated": {
+		// TODO: remove suffixes MostFreeCapacity/BestFit after dropping the TASMostFreeCapacity feature gate
+		"minimize the number of used racks before optimizing the number of nodes; MostFreeCapacity": {
 			// Solution by optimizing the number of racks then nodes: [r3]: [x3,x4,x5,x6]
 			// Solution by optimizing the number of nodes: [r1,r2]: [x1,x2]
 			//
@@ -348,9 +348,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"block required; 4 pods fit into one host each; LeastAllocated": {
+		"block required; 4 pods fit into one host each; MostFreeCapacity": {
 			nodes: binaryTreesNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -389,9 +389,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"block required; 4 pods fit into one host each; MostAllocated": {
+		"block required; 4 pods fit into one host each; BestFit": {
 			nodes: binaryTreesNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -430,10 +430,10 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: false,
+			enableTASMostFreeCapacity: false,
 		},
-		"host required; single Pod fits in the host; LeastAllocated": {
-			// TODO: remove after dropping the TASLeastAllocated feature gate
+		"host required; single Pod fits in the host; MostFreeCapacity": {
+			// TODO: remove after dropping the TASMostFreeCapacity feature gate
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(corev1.LabelHostname),
@@ -454,9 +454,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"host required; single Pod fits in the host; MostAllocated": {
+		"host required; single Pod fits in the host; BestFit": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(corev1.LabelHostname),
@@ -477,10 +477,10 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: false,
+			enableTASMostFreeCapacity: false,
 		},
-		"rack required; single Pod fits in a rack; LeastAllocated": {
-			// TODO: remove after dropping the TASLeastAllocated feature gate
+		"rack required; single Pod fits in a rack; MostFreeCapacity": {
+			// TODO: remove after dropping the TASMostFreeCapacity feature gate
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasRackLabel),
@@ -502,9 +502,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"rack required; multiple Pods fits in a rack; LeastAllocated": {
+		"rack required; multiple Pods fits in a rack; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasRackLabel),
@@ -526,9 +526,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"rack required; multiple Pods fit in a rack; MostAllocated": {
+		"rack required; multiple Pods fit in a rack; BestFit": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasRackLabel),
@@ -550,9 +550,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: false,
+			enableTASMostFreeCapacity: false,
 		},
-		"block preferred; Pods fit in 2 blocks; MostAllocated": {
+		"block preferred; Pods fit in 2 blocks; BestFit": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1").
 					Label(tasBlockLabel, "b1").
@@ -604,9 +604,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: false,
+			enableTASMostFreeCapacity: false,
 		},
-		"rack required; multiple Pods fit in some racks; MostAllocated": {
+		"rack required; multiple Pods fit in some racks; BestFit": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasRackLabel),
@@ -628,9 +628,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: false,
+			enableTASMostFreeCapacity: false,
 		},
-		"rack required; too many pods to fit in any rack; LeastAllocated": {
+		"rack required; too many pods to fit in any rack; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasRackLabel),
@@ -639,12 +639,12 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   4,
-			wantReason:              `topology "default" allows to fit only 3 out of 4 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     4,
+			wantReason:                `topology "default" allows to fit only 3 out of 4 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
-		"block required; single Pod fits in a block; LeastAllocated": {
-			// TODO: remove after dropping the TASLeastAllocated feature gate
+		"block required; single Pod fits in a block; MostFreeCapacity": {
+			// TODO: remove after dropping the TASMostFreeCapacity feature gate
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -669,9 +669,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"block required; single Pod fits in a block and a single rack; MostAllocated": {
+		"block required; single Pod fits in a block and a single rack; BestFit": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -696,9 +696,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: false,
+			enableTASMostFreeCapacity: false,
 		},
-		"block required; single Pod fits in a block spread across two racks; MostAllocated": {
+		"block required; single Pod fits in a block spread across two racks; BestFit": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -730,9 +730,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: false,
+			enableTASMostFreeCapacity: false,
 		},
-		"block required; Pods fit in a block spread across two racks; LeastAllocated": {
+		"block required; Pods fit in a block spread across two racks; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -761,9 +761,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"block required; single Pod which cannot be split; LeastAllocated": {
+		"block required; single Pod which cannot be split; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -772,11 +772,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 4000,
 			},
-			count:                   1,
-			wantReason:              `topology "default" doesn't allow to fit any of 1 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                `topology "default" doesn't allow to fit any of 1 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
-		"block required; too many Pods to fit requested; LeastAllocated": {
+		"block required; too many Pods to fit requested; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasBlockLabel),
@@ -785,12 +785,12 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   5,
-			wantReason:              `topology "default" allows to fit only 4 out of 5 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     5,
+			wantReason:                `topology "default" allows to fit only 4 out of 5 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
-		"rack required; single Pod requiring memory; LeastAllocated": {
-			// TODO: remove after dropping the TASLeastAllocated feature gate
+		"rack required; single Pod requiring memory; MostFreeCapacity": {
+			// TODO: remove after dropping the TASMostFreeCapacity feature gate
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Required: ptr.To(tasRackLabel),
@@ -812,9 +812,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"rack preferred; but only block can accommodate the workload; LeastAllocated": {
+		"rack preferred; but only block can accommodate the workload; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Preferred: ptr.To(tasRackLabel),
@@ -843,9 +843,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"rack preferred; but only multiple blocks can accommodate the workload; LeastAllocated": {
+		"rack preferred; but only multiple blocks can accommodate the workload; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Preferred: ptr.To(tasRackLabel),
@@ -881,9 +881,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"block preferred; but only multiple blocks can accommodate the workload; LeastAllocated": {
+		"block preferred; but only multiple blocks can accommodate the workload; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Preferred: ptr.To(tasBlockLabel),
@@ -919,9 +919,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"block preferred; but the workload cannot be accommodate in entire topology; LeastAllocated": {
+		"block preferred; but the workload cannot be accommodate in entire topology; MostFreeCapacity": {
 			nodes: defaultNodes,
 			topologyRequest: kueue.PodSetTopologyRequest{
 				Preferred: ptr.To(tasBlockLabel),
@@ -930,11 +930,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   10,
-			wantReason:              `topology "default" allows to fit only 7 out of 10 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     10,
+			wantReason:                `topology "default" allows to fit only 7 out of 10 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
-		"only nodes with matching labels are considered; no matching node; LeastAllocated": {
+		"only nodes with matching labels are considered; no matching node; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1-r1-x1").
 					Label("zone", "zone-a").
@@ -956,11 +956,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   1,
-			wantReason:              "no topology domains at level: kubernetes.io/hostname",
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                "no topology domains at level: kubernetes.io/hostname",
+			enableTASMostFreeCapacity: true,
 		},
-		"only nodes with matching labels are considered; matching node is found; LeastAllocated": {
+		"only nodes with matching labels are considered; matching node is found; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1-r1-x1").
 					Label("zone", "zone-a").
@@ -995,9 +995,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"only nodes with matching levels are considered; no host label on node; LeastAllocated": {
+		"only nodes with matching levels are considered; no host label on node; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1-r1-x1").
 					Label(tasBlockLabel, "b1").
@@ -1018,11 +1018,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   1,
-			wantReason:              "no topology domains at level: cloud.com/topology-rack",
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                "no topology domains at level: cloud.com/topology-rack",
+			enableTASMostFreeCapacity: true,
 		},
-		"don't consider unscheduled Pods when computing capacity; LeastAllocated": {
+		"don't consider unscheduled Pods when computing capacity; MostFreeCapacity": {
 			// the Pod is not scheduled (no NodeName set, so is not blocking capacity)
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("x1").
@@ -1059,9 +1059,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"don't consider terminal pods when computing the capacity; LeastAllocated": {
+		"don't consider terminal pods when computing the capacity; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("x1").
 					Label(corev1.LabelHostname, "x1").
@@ -1102,9 +1102,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"include usage from pending scheduled non-TAS pods, blocked assignment; LeastAllocated": {
+		"include usage from pending scheduled non-TAS pods, blocked assignment; MostFreeCapacity": {
 			// there is not enough free capacity on the only node x1
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("x1").
@@ -1130,11 +1130,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 600,
 			},
-			count:                   1,
-			wantReason:              `topology "default" doesn't allow to fit any of 1 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                `topology "default" doesn't allow to fit any of 1 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
-		"include usage from running non-TAS pods, blocked assignment; LeastAllocated": {
+		"include usage from running non-TAS pods, blocked assignment; MostFreeCapacity": {
 			// there is not enough free capacity on the only node x1
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("x1").
@@ -1160,11 +1160,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 600,
 			},
-			count:                   1,
-			wantReason:              `topology "default" doesn't allow to fit any of 1 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                `topology "default" doesn't allow to fit any of 1 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
-		"include usage from running non-TAS pods, found free capacity on another node; LeastAllocated": {
+		"include usage from running non-TAS pods, found free capacity on another node; MostFreeCapacity": {
 			// there is not enough free capacity on the node x1 as the
 			// assignments lends on the free x2
 			nodes: []corev1.Node{
@@ -1211,9 +1211,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"no assignment as node is not ready; LeastAllocated": {
+		"no assignment as node is not ready; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1-r1-x1").
 					Label("zone", "zone-a").
@@ -1240,11 +1240,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   1,
-			wantReason:              "no topology domains at level: kubernetes.io/hostname",
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                "no topology domains at level: kubernetes.io/hostname",
+			enableTASMostFreeCapacity: true,
 		},
-		"no assignment as node is unschedulable; LeastAllocated": {
+		"no assignment as node is unschedulable; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1-r1-x1").
 					Label("zone", "zone-a").
@@ -1268,11 +1268,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   1,
-			wantReason:              "no topology domains at level: kubernetes.io/hostname",
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                "no topology domains at level: kubernetes.io/hostname",
+			enableTASMostFreeCapacity: true,
 		},
-		"skip node which has untolerated taint; LeastAllocated": {
+		"skip node which has untolerated taint; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("x1").
 					Label("zone", "zone-a").
@@ -1300,11 +1300,11 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 1000,
 			},
-			count:                   1,
-			wantReason:              `topology "default" doesn't allow to fit any of 1 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                `topology "default" doesn't allow to fit any of 1 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
-		"allow to schedule on node with tolerated taint; LeastAllocated": {
+		"allow to schedule on node with tolerated taint; MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1-r1-x1").
 					Label("zone", "zone-a").
@@ -1351,9 +1351,9 @@ func TestFindTopologyAssignment(t *testing.T) {
 					},
 				},
 			},
-			enableTASLeastAllocated: true,
+			enableTASMostFreeCapacity: true,
 		},
-		"no assignment as node does not have enough allocatable pods (.status.allocatable['pods']); LeastAllocated": {
+		"no assignment as node does not have enough allocatable pods (.status.allocatable['pods']); MostFreeCapacity": {
 			nodes: []corev1.Node{
 				*testingnode.MakeNode("b1-r1-x1").
 					Label("zone", "zone-a").
@@ -1382,16 +1382,16 @@ func TestFindTopologyAssignment(t *testing.T) {
 			requests: resources.Requests{
 				corev1.ResourceCPU: 300,
 			},
-			count:                   1,
-			wantReason:              `topology "default" doesn't allow to fit any of 1 pod(s)`,
-			enableTASLeastAllocated: true,
+			count:                     1,
+			wantReason:                `topology "default" doesn't allow to fit any of 1 pod(s)`,
+			enableTASMostFreeCapacity: true,
 		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
 			ctx, _ := utiltesting.ContextWithLog(t)
-			// TODO: remove after dropping the TASLeastAllocated feature gate
-			features.SetFeatureGateDuringTest(t, features.TASLeastAllocated, tc.enableTASLeastAllocated)
+			// TODO: remove after dropping the TASMostFreeCapacity feature gate
+			features.SetFeatureGateDuringTest(t, features.TASMostFreeCapacity, tc.enableTASMostFreeCapacity)
 
 			initialObjects := make([]client.Object, 0)
 			for i := range tc.nodes {
diff --git a/pkg/controller/jobframework/reconciler_test.go b/pkg/controller/jobframework/reconciler_test.go
index 3a08bf5ac..d2599447b 100644
--- a/pkg/controller/jobframework/reconciler_test.go
+++ b/pkg/controller/jobframework/reconciler_test.go
@@ -101,7 +101,7 @@ func TestIsAncestorJobManaged(t *testing.T) {
 					Obj(),
 				testingmpijob.MakeMPIJob(parentJobName, jobNamespace).
 					UID(parentJobName).
-					OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind("AppWrapper")).
+					OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind(awv1beta2.AppWrapperKind)).
 					Obj(),
 			},
 			job: testingjob.MakeJob(childJobName, jobNamespace).
@@ -116,7 +116,7 @@ func TestIsAncestorJobManaged(t *testing.T) {
 					Obj(),
 				testingmpijob.MakeMPIJob(parentJobName, jobNamespace).
 					UID(parentJobName).
-					OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind("AppWrapper")).
+					OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind(awv1beta2.AppWrapperKind)).
 					Obj(),
 			},
 			job: testingjob.MakeJob(childJobName, jobNamespace).
@@ -132,7 +132,7 @@ func TestIsAncestorJobManaged(t *testing.T) {
 					Obj(),
 				testingmpijob.MakeMPIJob(parentJobName, jobNamespace).
 					UID(parentJobName).
-					OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind("AppWrapper")).
+					OwnerReference(grandparentJobName, awv1beta2.GroupVersion.WithKind(awv1beta2.AppWrapperKind)).
 					Obj(),
 			},
 			job: testingjob.MakeJob(childJobName, jobNamespace).
diff --git a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
index 5be078fce..1e61fe688 100644
--- a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
+++ b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
@@ -440,7 +440,7 @@ func TestValidateUpdate(t *testing.T) {
 			newObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
 				WithOwnerReference(metav1.OwnerReference{
 					APIVersion: awv1beta2.GroupVersion.String(),
-					Kind:       "AppWrapper",
+					Kind:       awv1beta2.AppWrapperKind,
 					Controller: ptr.To(true),
 				}).
 				Queue("test-queue").
@@ -457,7 +457,7 @@ func TestValidateUpdate(t *testing.T) {
 			newObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
 				WithOwnerReference(metav1.OwnerReference{
 					APIVersion: awv1beta2.GroupVersion.String(),
-					Kind:       "AppWrapper",
+					Kind:       awv1beta2.AppWrapperKind,
 					Controller: ptr.To(true),
 				}).
 				Queue("test-queue").
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 8d652d52e..65d3bc9cc 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -4285,6 +4285,55 @@ func TestScheduleForTAS(t *testing.T) {
 		// eventCmpOpts are the comparison options for the events
 		eventCmpOpts []cmp.Option
 	}{
+		"workload which does not specify TAS annotation uses the only TAS flavor": {
+			nodes:           defaultSingleNode,
+			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASFlavor, defaultFlavor},
+			clusterQueues: []kueue.ClusterQueue{
+				*utiltesting.MakeClusterQueue("tas-main").
+					ResourceGroup(
+						*utiltesting.MakeFlavorQuotas("tas-default").
+							Resource(corev1.ResourceCPU, "50").Obj()).
+					Obj(),
+			},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo", "default").
+					Queue("tas-main").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Obj(),
+			},
+			wantNewAssignments: map[string]kueue.Admission{
+				"default/foo": *utiltesting.MakeAdmission("tas-main", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "1000m").
+					AssignmentPodCount(1).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"x1",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			eventCmpOpts: []cmp.Option{eventIgnoreMessage},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "foo"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "foo"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
 		"workload requiring TAS skips the non-TAS flavor": {
 			nodes:           defaultSingleNode,
 			topologies:      []kueuealpha.Topology{defaultSingleLevelTopology},
diff --git a/test/integration/singlecluster/scheduler/scheduler_test.go b/test/integration/singlecluster/scheduler/scheduler_test.go
index 27aeb4cf9..0d204baaa 100644
--- a/test/integration/singlecluster/scheduler/scheduler_test.go
+++ b/test/integration/singlecluster/scheduler/scheduler_test.go
@@ -1503,8 +1503,8 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Create(ctx, queue)).Should(gomega.Succeed())
 
 			ginkgo.By("cohorts created")
-			cohortRight := testing.MakeCohort("left").Parent("root").Obj()
-			cohortLeft := testing.MakeCohort("right").Parent("root").Obj()
+			cohortLeft := testing.MakeCohort("left").Parent("root").Obj()
+			cohortRight := testing.MakeCohort("right").Parent("root").Obj()
 			gomega.Expect(k8sClient.Create(ctx, cohortLeft)).Should(gomega.Succeed())
 			gomega.Expect(k8sClient.Create(ctx, cohortRight)).Should(gomega.Succeed())
 
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index 18c20a584..cd5ed1de1 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -29,6 +29,7 @@ import (
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/features"
+	utiltas "sigs.k8s.io/kueue/pkg/util/tas"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	testingnode "sigs.k8s.io/kueue/pkg/util/testingjobs/node"
 	"sigs.k8s.io/kueue/test/util"
@@ -356,6 +357,16 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 				}
 			})
 
+			ginkgo.It("should expose the TopologyName in LocalQueue status", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdLocalQueue := &kueue.LocalQueue{}
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(localQueue), createdLocalQueue)).Should(gomega.Succeed())
+					g.Expect(createdLocalQueue.Status.Flavors).Should(gomega.HaveLen(1))
+					g.Expect(createdLocalQueue.Status.Flavors[0].Topology).ShouldNot(gomega.BeNil())
+					g.Expect(createdLocalQueue.Status.Flavors[0].Topology.Levels).Should(gomega.Equal(utiltas.Levels(topology)))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
 			ginkgo.It("should not admit workload which does not fit to the required topology domain", func() {
 				ginkgo.By("creating a workload which requires rack, but does not fit in any", func() {
 					wl1 := testing.MakeWorkload("wl1-inadmissible", ns.Name).
@@ -809,6 +820,49 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 1)
 				})
 			})
+
+			ginkgo.It("should admit workload using TAS without explicit TAS annotation", func() {
+				var wl1, wl2 *kueue.Workload
+				ginkgo.By("creating a workload without explicit TAS annotation which can fit", func() {
+					wl1 = testing.MakeWorkload("wl1", ns.Name).
+						PodSets(*testing.MakePodSet("worker", 2).
+							Obj()).
+						Queue(localQueue.Name).Request(corev1.ResourceCPU, "1").Obj()
+					gomega.Expect(k8sClient.Create(ctx, wl1)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the workload is admitted", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+					gomega.Expect(wl1.Status.Admission.PodSetAssignments[0].TopologyAssignment).Should(gomega.BeComparableTo(
+						&kueue.TopologyAssignment{
+							Levels: []string{corev1.LabelHostname},
+							Domains: []kueue.TopologyDomainAssignment{
+								{Count: 1, Values: []string{"x1"}},
+								{Count: 1, Values: []string{"x2"}},
+							},
+						},
+					))
+				})
+
+				ginkgo.By("creating the second workload without explicit TAS annotation which cannot fit", func() {
+					wl2 = testing.MakeWorkload("wl2", ns.Name).
+						PodSets(*testing.MakePodSet("worker-2", 3).
+							Obj()).
+						Queue(localQueue.Name).Request(corev1.ResourceCPU, "1").Obj()
+					gomega.Expect(k8sClient.Create(ctx, wl2)).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the wl2 has not been admitted", func() {
+					util.ExpectWorkloadsToBePending(ctx, k8sClient, wl2)
+					util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 1)
+				})
+
+				ginkgo.By("finish wl1 and verify wl2 can fit now", func() {
+					util.FinishWorkloads(ctx, k8sClient, wl1)
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl2)
+				})
+			})
 		})
 
 		ginkgo.When("Preemption is enabled within ClusterQueue", func() {
@@ -1330,6 +1384,18 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 				}
 			})
 
+			ginkgo.It("should expose the TopologyName in LocalQueue status", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdLocalQueue := &kueue.LocalQueue{}
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(localQueue), createdLocalQueue)).Should(gomega.Succeed())
+					g.Expect(createdLocalQueue.Status.Flavors).Should(gomega.HaveLen(2))
+					for _, flavor := range createdLocalQueue.Status.Flavors {
+						g.Expect(flavor.Topology).ShouldNot(gomega.BeNil())
+						g.Expect(flavor.Topology.Levels).Should(gomega.Equal(utiltas.Levels(topology)))
+					}
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
 			ginkgo.It("should admit workload which fits in a required topology domain", func() {
 				var wl1 *kueue.Workload
 				ginkgo.By("creating a workload which requires block and can fit", func() {
diff --git a/test/util/e2e.go b/test/util/e2e.go
index 4af8604fa..f75063d35 100644
--- a/test/util/e2e.go
+++ b/test/util/e2e.go
@@ -259,7 +259,10 @@ func WaitForActivePodsAndTerminate(ctx context.Context, k8sClient client.Client,
 		g.Expect(k8sClient.List(ctx, &pods, podListOpts)).To(gomega.Succeed())
 		activePods = make([]corev1.Pod, 0)
 		for _, p := range pods.Items {
-			if (len(p.Status.PodIP) != 0 && p.Status.PodIP != "0.0.0.0") && (p.Status.Phase == corev1.PodRunning || p.Status.Phase == corev1.PodPending) {
+			if len(p.Status.PodIP) != 0 && p.Status.Phase == corev1.PodRunning {
+				cmd := []string{"/bin/sh", "-c", fmt.Sprintf("curl \"http://%s:8080/readyz\"", p.Status.PodIP)}
+				_, _, err := KExecute(ctx, cfg, restClient, namespace, p.Name, p.Spec.Containers[0].Name, cmd)
+				g.Expect(err).ToNot(gomega.HaveOccurred())
 				activePods = append(activePods, p)
 			}
 		}