Repository: kubernetes-sigs/kueue

Issue:
TAS: Two-level JobSet scheduling
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:
A possibility to define topology for each Job in ReplicatedJob to co-locate the whole ReplicatedJob in one domain and each Job in that ReplicatedJob in lower-level domain within that domain.

In more generic terms - an API to define required sub PodSet topology and allow for two-level scheduling for the whole PodSet and lower-level topology domain for each chunk within that PodSet.

An example is a JobSet in which user can define topology for all Jobs resulting from a single ReplicatedJob. We want to allow user to define topology for each such Job and possibly co-locate all such Jobs in higher-level topology.

**Why is this needed**:
Currently Topology-Aware Scheduling considers the whole PodSet as a single unit for which Kueue looks for a matching domain. For JobSet PodSet covers all pods resulting from the whole ReplicatedJob. However, there are use-cases in which user would like to co-locate each Job in a lower-level domains and co-locate all those Jobs in higher-level domain.

Additionally, user would like to define a topology only for each Job without any expectations for a topology of the whole ReplicatedJob.

**Completion requirements**:

This enhancement requires the following artifacts:

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

The artifacts should be linked in subsequent comments.


Hints from issue comments:
@mimowo: /retitle TAS: Two-level JobSet scheduling@lchrzaszcz: TAS KEP update to reflect two-level scheduling and podset chunk topology: https://github.com/kubernetes-sigs/kueue/pull/5449@kerthcet: One quick question: can this solve the https://github.com/kubernetes-sigs/kueue/pull/3941? E.g. co-locate the master and worker jobs in pytorchjob.
Thanks!@kerthcet: I'll read the KEP in detail later.@kerthcet: This focused on podSet chunks, not between multi-podsets. @lchrzaszcz: I will be looking into cross-podsets scheduling in few following weeks. I'll do it mostly for LeaderWorkerSet though, but I think the underlying mechanism will be universal, so probably it'll be easy to add pytorchjob support afterwards.@kerthcet: Thanks @lchrzaszcz for your response, I think leaderworkerset pytorchjob are just two different workload types, the mechanism should be universal to them all I believe. Glad to jump into the discussion.@tenzen-y: /assign @lchrzaszcz 

Solution patch:
diff --git a/apis/kueue/v1alpha1/cohort_types.go b/apis/kueue/v1alpha1/cohort_types.go
index 778c06e93..6a614341e 100644
--- a/apis/kueue/v1alpha1/cohort_types.go
+++ b/apis/kueue/v1alpha1/cohort_types.go
@@ -73,6 +73,7 @@ type CohortStatus struct {
 	FairSharing *kueuebeta.FairSharingStatus `json:"fairSharing,omitempty"`
 }
 
+//+genclient
 //+kubebuilder:object:root=true
 //+kubebuilder:resource:scope=Cluster
 //+kubebuilder:subresource:status
diff --git a/apis/kueue/v1alpha1/topology_types.go b/apis/kueue/v1alpha1/topology_types.go
index e8fa7ac9d..316b52d0a 100644
--- a/apis/kueue/v1alpha1/topology_types.go
+++ b/apis/kueue/v1alpha1/topology_types.go
@@ -46,6 +46,19 @@ const (
 	// +kubebuilder:validation:Type=boolean
 	PodSetUnconstrainedTopologyAnnotation = "kueue.x-k8s.io/podset-unconstrained-topology"
 
+	// PodSetSliceRequiredTopologyAnnotation indicates that a PodSet requires
+	// Topology Aware Scheduling, and requires scheduling each PodSet slice on nodes
+	// within the topology domain corresponding to the topology level
+	// indicated by the annotation value (e.g. within a rack or within a block).
+	PodSetSliceRequiredTopologyAnnotation = "kueue.x-k8s.io/podset-slice-required-topology"
+
+	// PodSetSliceSizeAnnotation describes the requested size of a podset slice
+	// for which Kueue finds a requested topology domain.
+	//
+	// This annotation is required if `kueue.x-k8s.io/podset-slice-required-topology`
+	// is defined.
+	PodSetSliceSizeAnnotation = "kueue.x-k8s.io/podset-slice-size"
+
 	// TopologySchedulingGate is used to delay scheduling of a Pod until the
 	// nodeSelectors corresponding to the assigned topology domain are injected
 	// into the Pod. For the Pod-based integrations the gate is added in webhook
diff --git a/apis/kueue/v1beta1/workload_types.go b/apis/kueue/v1beta1/workload_types.go
index 0b5c19794..7f8aedb57 100644
--- a/apis/kueue/v1beta1/workload_types.go
+++ b/apis/kueue/v1beta1/workload_types.go
@@ -124,6 +124,19 @@ type PodSetTopologyRequest struct {
 	// SubGroupIndexLabel indicates the count of replicated Jobs (groups) within a PodSet.
 	// For example, in the context of JobSet this value is read from jobset.sigs.k8s.io/replicatedjob-replicas.
 	SubGroupCount *int32 `json:"subGroupCount,omitempty"`
+
+	// PodSetSliceRequiredTopology indicates the topology level required by the PodSet slice, as
+	// indicated by the `kueue.x-k8s.io/podset-slice-required-topology` annotation.
+	//
+	// +optional
+	PodSetSliceRequiredTopology *string `json:"podSetSliceRequiredTopology,omitempty"`
+
+	// PodSetSliceSize indicates the size of a subgroup of pods in a PodSet for which
+	// Kueue finds a requested topology domain on a level defined
+	// in `kueue.x-k8s.io/podset-slice-required-topology` annotation.
+	//
+	// +optional
+	PodSetSliceSize *int32 `json:"podSetSliceSize,omitempty"`
 }
 
 type Admission struct {
diff --git a/apis/kueue/v1beta1/zz_generated.deepcopy.go b/apis/kueue/v1beta1/zz_generated.deepcopy.go
index 492abd3c9..78890789a 100644
--- a/apis/kueue/v1beta1/zz_generated.deepcopy.go
+++ b/apis/kueue/v1beta1/zz_generated.deepcopy.go
@@ -1138,6 +1138,16 @@ func (in *PodSetTopologyRequest) DeepCopyInto(out *PodSetTopologyRequest) {
 		*out = new(int32)
 		**out = **in
 	}
+	if in.PodSetSliceRequiredTopology != nil {
+		in, out := &in.PodSetSliceRequiredTopology, &out.PodSetSliceRequiredTopology
+		*out = new(string)
+		**out = **in
+	}
+	if in.PodSetSliceSize != nil {
+		in, out := &in.PodSetSliceSize, &out.PodSetSliceSize
+		*out = new(int32)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PodSetTopologyRequest.
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
index c00943ce9..acb06954b 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
@@ -7833,6 +7833,18 @@ spec:
                               - JobSet: kubernetes.io/job-completion-index (inherited from Job)
                               - Kubeflow: training.kubeflow.org/replica-index
                             type: string
+                          podSetSliceRequiredTopology:
+                            description: |-
+                              PodSetSliceRequiredTopology indicates the topology level required by the PodSet slice, as
+                              indicated by the `kueue.x-k8s.io/podset-slice-required-topology` annotation.
+                            type: string
+                          podSetSliceSize:
+                            description: |-
+                              PodSetSliceSize indicates the size of a subgroup of pods in a PodSet for which
+                              Kueue finds a requested topology domain on a level defined
+                              in `kueue.x-k8s.io/podset-slice-required-topology` annotation.
+                            format: int32
+                            type: integer
                           preferred:
                             description: |-
                               preferred indicates the topology level preferred by the PodSet, as
diff --git a/client-go/applyconfiguration/kueue/v1alpha1/cohort.go b/client-go/applyconfiguration/kueue/v1alpha1/cohort.go
new file mode 100644
index 000000000..69ab523ef
--- /dev/null
+++ b/client-go/applyconfiguration/kueue/v1alpha1/cohort.go
@@ -0,0 +1,224 @@
+/*
+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 v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	v1 "k8s.io/client-go/applyconfigurations/meta/v1"
+)
+
+// CohortApplyConfiguration represents a declarative configuration of the Cohort type for use
+// with apply.
+type CohortApplyConfiguration struct {
+	v1.TypeMetaApplyConfiguration    `json:",inline"`
+	*v1.ObjectMetaApplyConfiguration `json:"metadata,omitempty"`
+	Spec                             *CohortSpecApplyConfiguration   `json:"spec,omitempty"`
+	Status                           *CohortStatusApplyConfiguration `json:"status,omitempty"`
+}
+
+// Cohort constructs a declarative configuration of the Cohort type for use with
+// apply.
+func Cohort(name, namespace string) *CohortApplyConfiguration {
+	b := &CohortApplyConfiguration{}
+	b.WithName(name)
+	b.WithNamespace(namespace)
+	b.WithKind("Cohort")
+	b.WithAPIVersion("kueue.x-k8s.io/v1alpha1")
+	return b
+}
+
+// WithKind sets the Kind 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 Kind field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithKind(value string) *CohortApplyConfiguration {
+	b.TypeMetaApplyConfiguration.Kind = &value
+	return b
+}
+
+// WithAPIVersion sets the APIVersion 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 APIVersion field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithAPIVersion(value string) *CohortApplyConfiguration {
+	b.TypeMetaApplyConfiguration.APIVersion = &value
+	return b
+}
+
+// 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 *CohortApplyConfiguration) WithName(value string) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.Name = &value
+	return b
+}
+
+// WithGenerateName sets the GenerateName 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 GenerateName field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithGenerateName(value string) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.GenerateName = &value
+	return b
+}
+
+// WithNamespace sets the Namespace 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 Namespace field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithNamespace(value string) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.Namespace = &value
+	return b
+}
+
+// WithUID sets the UID 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 UID field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithUID(value types.UID) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.UID = &value
+	return b
+}
+
+// WithResourceVersion sets the ResourceVersion 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 ResourceVersion field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithResourceVersion(value string) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.ResourceVersion = &value
+	return b
+}
+
+// WithGeneration sets the Generation 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 Generation field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithGeneration(value int64) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.Generation = &value
+	return b
+}
+
+// WithCreationTimestamp sets the CreationTimestamp 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 CreationTimestamp field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithCreationTimestamp(value metav1.Time) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.CreationTimestamp = &value
+	return b
+}
+
+// WithDeletionTimestamp sets the DeletionTimestamp 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 DeletionTimestamp field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithDeletionTimestamp(value metav1.Time) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.DeletionTimestamp = &value
+	return b
+}
+
+// WithDeletionGracePeriodSeconds sets the DeletionGracePeriodSeconds 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 DeletionGracePeriodSeconds field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithDeletionGracePeriodSeconds(value int64) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	b.ObjectMetaApplyConfiguration.DeletionGracePeriodSeconds = &value
+	return b
+}
+
+// WithLabels puts the entries into the Labels field in the declarative configuration
+// and returns the receiver, so that objects can be build by chaining "With" function invocations.
+// If called multiple times, the entries provided by each call will be put on the Labels field,
+// overwriting an existing map entries in Labels field with the same key.
+func (b *CohortApplyConfiguration) WithLabels(entries map[string]string) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	if b.ObjectMetaApplyConfiguration.Labels == nil && len(entries) > 0 {
+		b.ObjectMetaApplyConfiguration.Labels = make(map[string]string, len(entries))
+	}
+	for k, v := range entries {
+		b.ObjectMetaApplyConfiguration.Labels[k] = v
+	}
+	return b
+}
+
+// WithAnnotations puts the entries into the Annotations field in the declarative configuration
+// and returns the receiver, so that objects can be build by chaining "With" function invocations.
+// If called multiple times, the entries provided by each call will be put on the Annotations field,
+// overwriting an existing map entries in Annotations field with the same key.
+func (b *CohortApplyConfiguration) WithAnnotations(entries map[string]string) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	if b.ObjectMetaApplyConfiguration.Annotations == nil && len(entries) > 0 {
+		b.ObjectMetaApplyConfiguration.Annotations = make(map[string]string, len(entries))
+	}
+	for k, v := range entries {
+		b.ObjectMetaApplyConfiguration.Annotations[k] = v
+	}
+	return b
+}
+
+// WithOwnerReferences adds the given value to the OwnerReferences 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 OwnerReferences field.
+func (b *CohortApplyConfiguration) WithOwnerReferences(values ...*v1.OwnerReferenceApplyConfiguration) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	for i := range values {
+		if values[i] == nil {
+			panic("nil value passed to WithOwnerReferences")
+		}
+		b.ObjectMetaApplyConfiguration.OwnerReferences = append(b.ObjectMetaApplyConfiguration.OwnerReferences, *values[i])
+	}
+	return b
+}
+
+// WithFinalizers adds the given value to the Finalizers 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 Finalizers field.
+func (b *CohortApplyConfiguration) WithFinalizers(values ...string) *CohortApplyConfiguration {
+	b.ensureObjectMetaApplyConfigurationExists()
+	for i := range values {
+		b.ObjectMetaApplyConfiguration.Finalizers = append(b.ObjectMetaApplyConfiguration.Finalizers, values[i])
+	}
+	return b
+}
+
+func (b *CohortApplyConfiguration) ensureObjectMetaApplyConfigurationExists() {
+	if b.ObjectMetaApplyConfiguration == nil {
+		b.ObjectMetaApplyConfiguration = &v1.ObjectMetaApplyConfiguration{}
+	}
+}
+
+// WithSpec sets the Spec 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 Spec field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithSpec(value *CohortSpecApplyConfiguration) *CohortApplyConfiguration {
+	b.Spec = value
+	return b
+}
+
+// WithStatus sets the Status 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 Status field is set to the value of the last call.
+func (b *CohortApplyConfiguration) WithStatus(value *CohortStatusApplyConfiguration) *CohortApplyConfiguration {
+	b.Status = value
+	return b
+}
+
+// GetName retrieves the value of the Name field in the declarative configuration.
+func (b *CohortApplyConfiguration) GetName() *string {
+	b.ensureObjectMetaApplyConfigurationExists()
+	return b.ObjectMetaApplyConfiguration.Name
+}
diff --git a/client-go/applyconfiguration/kueue/v1alpha1/cohortspec.go b/client-go/applyconfiguration/kueue/v1alpha1/cohortspec.go
new file mode 100644
index 000000000..3320d7d9f
--- /dev/null
+++ b/client-go/applyconfiguration/kueue/v1alpha1/cohortspec.go
@@ -0,0 +1,66 @@
+/*
+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 v1alpha1
+
+import (
+	v1beta1 "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	kueuev1beta1 "sigs.k8s.io/kueue/client-go/applyconfiguration/kueue/v1beta1"
+)
+
+// CohortSpecApplyConfiguration represents a declarative configuration of the CohortSpec type for use
+// with apply.
+type CohortSpecApplyConfiguration struct {
+	Parent         *v1beta1.CohortReference                       `json:"parent,omitempty"`
+	ResourceGroups []kueuev1beta1.ResourceGroupApplyConfiguration `json:"resourceGroups,omitempty"`
+	FairSharing    *kueuev1beta1.FairSharingApplyConfiguration    `json:"fairSharing,omitempty"`
+}
+
+// CohortSpecApplyConfiguration constructs a declarative configuration of the CohortSpec type for use with
+// apply.
+func CohortSpec() *CohortSpecApplyConfiguration {
+	return &CohortSpecApplyConfiguration{}
+}
+
+// WithParent sets the Parent 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 Parent field is set to the value of the last call.
+func (b *CohortSpecApplyConfiguration) WithParent(value v1beta1.CohortReference) *CohortSpecApplyConfiguration {
+	b.Parent = &value
+	return b
+}
+
+// WithResourceGroups adds the given value to the ResourceGroups 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 ResourceGroups field.
+func (b *CohortSpecApplyConfiguration) WithResourceGroups(values ...*kueuev1beta1.ResourceGroupApplyConfiguration) *CohortSpecApplyConfiguration {
+	for i := range values {
+		if values[i] == nil {
+			panic("nil value passed to WithResourceGroups")
+		}
+		b.ResourceGroups = append(b.ResourceGroups, *values[i])
+	}
+	return b
+}
+
+// WithFairSharing sets the FairSharing 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 FairSharing field is set to the value of the last call.
+func (b *CohortSpecApplyConfiguration) WithFairSharing(value *kueuev1beta1.FairSharingApplyConfiguration) *CohortSpecApplyConfiguration {
+	b.FairSharing = value
+	return b
+}
diff --git a/client-go/applyconfiguration/kueue/v1alpha1/cohortstatus.go b/client-go/applyconfiguration/kueue/v1alpha1/cohortstatus.go
new file mode 100644
index 000000000..875f84f08
--- /dev/null
+++ b/client-go/applyconfiguration/kueue/v1alpha1/cohortstatus.go
@@ -0,0 +1,42 @@
+/*
+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 v1alpha1
+
+import (
+	v1beta1 "sigs.k8s.io/kueue/client-go/applyconfiguration/kueue/v1beta1"
+)
+
+// CohortStatusApplyConfiguration represents a declarative configuration of the CohortStatus type for use
+// with apply.
+type CohortStatusApplyConfiguration struct {
+	FairSharing *v1beta1.FairSharingStatusApplyConfiguration `json:"fairSharing,omitempty"`
+}
+
+// CohortStatusApplyConfiguration constructs a declarative configuration of the CohortStatus type for use with
+// apply.
+func CohortStatus() *CohortStatusApplyConfiguration {
+	return &CohortStatusApplyConfiguration{}
+}
+
+// WithFairSharing sets the FairSharing 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 FairSharing field is set to the value of the last call.
+func (b *CohortStatusApplyConfiguration) WithFairSharing(value *v1beta1.FairSharingStatusApplyConfiguration) *CohortStatusApplyConfiguration {
+	b.FairSharing = value
+	return b
+}
diff --git a/client-go/applyconfiguration/kueue/v1beta1/podsettopologyrequest.go b/client-go/applyconfiguration/kueue/v1beta1/podsettopologyrequest.go
index 3756f7181..4c5a70b3d 100644
--- a/client-go/applyconfiguration/kueue/v1beta1/podsettopologyrequest.go
+++ b/client-go/applyconfiguration/kueue/v1beta1/podsettopologyrequest.go
@@ -20,12 +20,14 @@ package v1beta1
 // PodSetTopologyRequestApplyConfiguration represents a declarative configuration of the PodSetTopologyRequest type for use
 // with apply.
 type PodSetTopologyRequestApplyConfiguration struct {
-	Required           *string `json:"required,omitempty"`
-	Preferred          *string `json:"preferred,omitempty"`
-	Unconstrained      *bool   `json:"unconstrained,omitempty"`
-	PodIndexLabel      *string `json:"podIndexLabel,omitempty"`
-	SubGroupIndexLabel *string `json:"subGroupIndexLabel,omitempty"`
-	SubGroupCount      *int32  `json:"subGroupCount,omitempty"`
+	Required                    *string `json:"required,omitempty"`
+	Preferred                   *string `json:"preferred,omitempty"`
+	Unconstrained               *bool   `json:"unconstrained,omitempty"`
+	PodIndexLabel               *string `json:"podIndexLabel,omitempty"`
+	SubGroupIndexLabel          *string `json:"subGroupIndexLabel,omitempty"`
+	SubGroupCount               *int32  `json:"subGroupCount,omitempty"`
+	PodSetSliceRequiredTopology *string `json:"podSetSliceRequiredTopology,omitempty"`
+	PodSetSliceSize             *int32  `json:"podSetSliceSize,omitempty"`
 }
 
 // PodSetTopologyRequestApplyConfiguration constructs a declarative configuration of the PodSetTopologyRequest type for use with
@@ -81,3 +83,19 @@ func (b *PodSetTopologyRequestApplyConfiguration) WithSubGroupCount(value int32)
 	b.SubGroupCount = &value
 	return b
 }
+
+// WithPodSetSliceRequiredTopology sets the PodSetSliceRequiredTopology 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 PodSetSliceRequiredTopology field is set to the value of the last call.
+func (b *PodSetTopologyRequestApplyConfiguration) WithPodSetSliceRequiredTopology(value string) *PodSetTopologyRequestApplyConfiguration {
+	b.PodSetSliceRequiredTopology = &value
+	return b
+}
+
+// WithPodSetSliceSize sets the PodSetSliceSize 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 PodSetSliceSize field is set to the value of the last call.
+func (b *PodSetTopologyRequestApplyConfiguration) WithPodSetSliceSize(value int32) *PodSetTopologyRequestApplyConfiguration {
+	b.PodSetSliceSize = &value
+	return b
+}
diff --git a/client-go/applyconfiguration/utils.go b/client-go/applyconfiguration/utils.go
index fe6c59a4d..c610295d0 100644
--- a/client-go/applyconfiguration/utils.go
+++ b/client-go/applyconfiguration/utils.go
@@ -35,6 +35,12 @@ import (
 func ForKind(kind schema.GroupVersionKind) interface{} {
 	switch kind {
 	// Group=kueue.x-k8s.io, Version=v1alpha1
+	case v1alpha1.SchemeGroupVersion.WithKind("Cohort"):
+		return &kueuev1alpha1.CohortApplyConfiguration{}
+	case v1alpha1.SchemeGroupVersion.WithKind("CohortSpec"):
+		return &kueuev1alpha1.CohortSpecApplyConfiguration{}
+	case v1alpha1.SchemeGroupVersion.WithKind("CohortStatus"):
+		return &kueuev1alpha1.CohortStatusApplyConfiguration{}
 	case v1alpha1.SchemeGroupVersion.WithKind("Topology"):
 		return &kueuev1alpha1.TopologyApplyConfiguration{}
 	case v1alpha1.SchemeGroupVersion.WithKind("TopologyLevel"):
diff --git a/client-go/clientset/versioned/typed/kueue/v1alpha1/cohort.go b/client-go/clientset/versioned/typed/kueue/v1alpha1/cohort.go
new file mode 100644
index 000000000..ce7858f2e
--- /dev/null
+++ b/client-go/clientset/versioned/typed/kueue/v1alpha1/cohort.go
@@ -0,0 +1,73 @@
+/*
+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 client-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	gentype "k8s.io/client-go/gentype"
+	kueuev1alpha1 "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	applyconfigurationkueuev1alpha1 "sigs.k8s.io/kueue/client-go/applyconfiguration/kueue/v1alpha1"
+	scheme "sigs.k8s.io/kueue/client-go/clientset/versioned/scheme"
+)
+
+// CohortsGetter has a method to return a CohortInterface.
+// A group's client should implement this interface.
+type CohortsGetter interface {
+	Cohorts(namespace string) CohortInterface
+}
+
+// CohortInterface has methods to work with Cohort resources.
+type CohortInterface interface {
+	Create(ctx context.Context, cohort *kueuev1alpha1.Cohort, opts v1.CreateOptions) (*kueuev1alpha1.Cohort, error)
+	Update(ctx context.Context, cohort *kueuev1alpha1.Cohort, opts v1.UpdateOptions) (*kueuev1alpha1.Cohort, error)
+	// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+	UpdateStatus(ctx context.Context, cohort *kueuev1alpha1.Cohort, opts v1.UpdateOptions) (*kueuev1alpha1.Cohort, error)
+	Delete(ctx context.Context, name string, opts v1.DeleteOptions) error
+	DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error
+	Get(ctx context.Context, name string, opts v1.GetOptions) (*kueuev1alpha1.Cohort, error)
+	List(ctx context.Context, opts v1.ListOptions) (*kueuev1alpha1.CohortList, error)
+	Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error)
+	Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *kueuev1alpha1.Cohort, err error)
+	Apply(ctx context.Context, cohort *applyconfigurationkueuev1alpha1.CohortApplyConfiguration, opts v1.ApplyOptions) (result *kueuev1alpha1.Cohort, err error)
+	// Add a +genclient:noStatus comment above the type to avoid generating ApplyStatus().
+	ApplyStatus(ctx context.Context, cohort *applyconfigurationkueuev1alpha1.CohortApplyConfiguration, opts v1.ApplyOptions) (result *kueuev1alpha1.Cohort, err error)
+	CohortExpansion
+}
+
+// cohorts implements CohortInterface
+type cohorts struct {
+	*gentype.ClientWithListAndApply[*kueuev1alpha1.Cohort, *kueuev1alpha1.CohortList, *applyconfigurationkueuev1alpha1.CohortApplyConfiguration]
+}
+
+// newCohorts returns a Cohorts
+func newCohorts(c *KueueV1alpha1Client, namespace string) *cohorts {
+	return &cohorts{
+		gentype.NewClientWithListAndApply[*kueuev1alpha1.Cohort, *kueuev1alpha1.CohortList, *applyconfigurationkueuev1alpha1.CohortApplyConfiguration](
+			"cohorts",
+			c.RESTClient(),
+			scheme.ParameterCodec,
+			namespace,
+			func() *kueuev1alpha1.Cohort { return &kueuev1alpha1.Cohort{} },
+			func() *kueuev1alpha1.CohortList { return &kueuev1alpha1.CohortList{} },
+		),
+	}
+}
diff --git a/client-go/clientset/versioned/typed/kueue/v1alpha1/fake/fake_cohort.go b/client-go/clientset/versioned/typed/kueue/v1alpha1/fake/fake_cohort.go
new file mode 100644
index 000000000..73858224a
--- /dev/null
+++ b/client-go/clientset/versioned/typed/kueue/v1alpha1/fake/fake_cohort.go
@@ -0,0 +1,50 @@
+/*
+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 client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	gentype "k8s.io/client-go/gentype"
+	v1alpha1 "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	kueuev1alpha1 "sigs.k8s.io/kueue/client-go/applyconfiguration/kueue/v1alpha1"
+	typedkueuev1alpha1 "sigs.k8s.io/kueue/client-go/clientset/versioned/typed/kueue/v1alpha1"
+)
+
+// fakeCohorts implements CohortInterface
+type fakeCohorts struct {
+	*gentype.FakeClientWithListAndApply[*v1alpha1.Cohort, *v1alpha1.CohortList, *kueuev1alpha1.CohortApplyConfiguration]
+	Fake *FakeKueueV1alpha1
+}
+
+func newFakeCohorts(fake *FakeKueueV1alpha1, namespace string) typedkueuev1alpha1.CohortInterface {
+	return &fakeCohorts{
+		gentype.NewFakeClientWithListAndApply[*v1alpha1.Cohort, *v1alpha1.CohortList, *kueuev1alpha1.CohortApplyConfiguration](
+			fake.Fake,
+			namespace,
+			v1alpha1.SchemeGroupVersion.WithResource("cohorts"),
+			v1alpha1.SchemeGroupVersion.WithKind("Cohort"),
+			func() *v1alpha1.Cohort { return &v1alpha1.Cohort{} },
+			func() *v1alpha1.CohortList { return &v1alpha1.CohortList{} },
+			func(dst, src *v1alpha1.CohortList) { dst.ListMeta = src.ListMeta },
+			func(list *v1alpha1.CohortList) []*v1alpha1.Cohort { return gentype.ToPointerSlice(list.Items) },
+			func(list *v1alpha1.CohortList, items []*v1alpha1.Cohort) {
+				list.Items = gentype.FromPointerSlice(items)
+			},
+		),
+		fake,
+	}
+}
diff --git a/client-go/clientset/versioned/typed/kueue/v1alpha1/fake/fake_kueue_client.go b/client-go/clientset/versioned/typed/kueue/v1alpha1/fake/fake_kueue_client.go
index c06a43ff2..27d0b9b31 100644
--- a/client-go/clientset/versioned/typed/kueue/v1alpha1/fake/fake_kueue_client.go
+++ b/client-go/clientset/versioned/typed/kueue/v1alpha1/fake/fake_kueue_client.go
@@ -27,6 +27,10 @@ type FakeKueueV1alpha1 struct {
 	*testing.Fake
 }
 
+func (c *FakeKueueV1alpha1) Cohorts(namespace string) v1alpha1.CohortInterface {
+	return newFakeCohorts(c, namespace)
+}
+
 func (c *FakeKueueV1alpha1) Topologies() v1alpha1.TopologyInterface {
 	return newFakeTopologies(c)
 }
diff --git a/client-go/clientset/versioned/typed/kueue/v1alpha1/generated_expansion.go b/client-go/clientset/versioned/typed/kueue/v1alpha1/generated_expansion.go
index 6601341f1..2593a3544 100644
--- a/client-go/clientset/versioned/typed/kueue/v1alpha1/generated_expansion.go
+++ b/client-go/clientset/versioned/typed/kueue/v1alpha1/generated_expansion.go
@@ -17,4 +17,6 @@ limitations under the License.
 
 package v1alpha1
 
+type CohortExpansion interface{}
+
 type TopologyExpansion interface{}
diff --git a/client-go/clientset/versioned/typed/kueue/v1alpha1/kueue_client.go b/client-go/clientset/versioned/typed/kueue/v1alpha1/kueue_client.go
index b5f9ecdd8..3a6e7c02a 100644
--- a/client-go/clientset/versioned/typed/kueue/v1alpha1/kueue_client.go
+++ b/client-go/clientset/versioned/typed/kueue/v1alpha1/kueue_client.go
@@ -27,6 +27,7 @@ import (
 
 type KueueV1alpha1Interface interface {
 	RESTClient() rest.Interface
+	CohortsGetter
 	TopologiesGetter
 }
 
@@ -35,6 +36,10 @@ type KueueV1alpha1Client struct {
 	restClient rest.Interface
 }
 
+func (c *KueueV1alpha1Client) Cohorts(namespace string) CohortInterface {
+	return newCohorts(c, namespace)
+}
+
 func (c *KueueV1alpha1Client) Topologies() TopologyInterface {
 	return newTopologies(c)
 }
diff --git a/client-go/informers/externalversions/generic.go b/client-go/informers/externalversions/generic.go
index 8e3fa5013..ebbf7e4eb 100644
--- a/client-go/informers/externalversions/generic.go
+++ b/client-go/informers/externalversions/generic.go
@@ -54,6 +54,8 @@ func (f *genericInformer) Lister() cache.GenericLister {
 func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource) (GenericInformer, error) {
 	switch resource {
 	// Group=kueue.x-k8s.io, Version=v1alpha1
+	case v1alpha1.SchemeGroupVersion.WithResource("cohorts"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Kueue().V1alpha1().Cohorts().Informer()}, nil
 	case v1alpha1.SchemeGroupVersion.WithResource("topologies"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Kueue().V1alpha1().Topologies().Informer()}, nil
 
diff --git a/client-go/informers/externalversions/kueue/v1alpha1/cohort.go b/client-go/informers/externalversions/kueue/v1alpha1/cohort.go
new file mode 100644
index 000000000..ed3c4ac5b
--- /dev/null
+++ b/client-go/informers/externalversions/kueue/v1alpha1/cohort.go
@@ -0,0 +1,101 @@
+/*
+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 informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	context "context"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	apiskueuev1alpha1 "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	versioned "sigs.k8s.io/kueue/client-go/clientset/versioned"
+	internalinterfaces "sigs.k8s.io/kueue/client-go/informers/externalversions/internalinterfaces"
+	kueuev1alpha1 "sigs.k8s.io/kueue/client-go/listers/kueue/v1alpha1"
+)
+
+// CohortInformer provides access to a shared informer and lister for
+// Cohorts.
+type CohortInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() kueuev1alpha1.CohortLister
+}
+
+type cohortInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewCohortInformer constructs a new informer for Cohort type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewCohortInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredCohortInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredCohortInformer constructs a new informer for Cohort type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredCohortInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KueueV1alpha1().Cohorts(namespace).List(context.Background(), options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KueueV1alpha1().Cohorts(namespace).Watch(context.Background(), options)
+			},
+			ListWithContextFunc: func(ctx context.Context, options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KueueV1alpha1().Cohorts(namespace).List(ctx, options)
+			},
+			WatchFuncWithContext: func(ctx context.Context, options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KueueV1alpha1().Cohorts(namespace).Watch(ctx, options)
+			},
+		},
+		&apiskueuev1alpha1.Cohort{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *cohortInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredCohortInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *cohortInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&apiskueuev1alpha1.Cohort{}, f.defaultInformer)
+}
+
+func (f *cohortInformer) Lister() kueuev1alpha1.CohortLister {
+	return kueuev1alpha1.NewCohortLister(f.Informer().GetIndexer())
+}
diff --git a/client-go/informers/externalversions/kueue/v1alpha1/interface.go b/client-go/informers/externalversions/kueue/v1alpha1/interface.go
index aebd35f5b..e8594fde7 100644
--- a/client-go/informers/externalversions/kueue/v1alpha1/interface.go
+++ b/client-go/informers/externalversions/kueue/v1alpha1/interface.go
@@ -23,6 +23,8 @@ import (
 
 // Interface provides access to all the informers in this group version.
 type Interface interface {
+	// Cohorts returns a CohortInformer.
+	Cohorts() CohortInformer
 	// Topologies returns a TopologyInformer.
 	Topologies() TopologyInformer
 }
@@ -38,6 +40,11 @@ func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakList
 	return &version{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
 }
 
+// Cohorts returns a CohortInformer.
+func (v *version) Cohorts() CohortInformer {
+	return &cohortInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
+
 // Topologies returns a TopologyInformer.
 func (v *version) Topologies() TopologyInformer {
 	return &topologyInformer{factory: v.factory, tweakListOptions: v.tweakListOptions}
diff --git a/client-go/listers/kueue/v1alpha1/cohort.go b/client-go/listers/kueue/v1alpha1/cohort.go
new file mode 100644
index 000000000..e002bdf74
--- /dev/null
+++ b/client-go/listers/kueue/v1alpha1/cohort.go
@@ -0,0 +1,69 @@
+/*
+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 lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	labels "k8s.io/apimachinery/pkg/labels"
+	listers "k8s.io/client-go/listers"
+	cache "k8s.io/client-go/tools/cache"
+	kueuev1alpha1 "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+)
+
+// CohortLister helps list Cohorts.
+// All objects returned here must be treated as read-only.
+type CohortLister interface {
+	// List lists all Cohorts in the indexer.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*kueuev1alpha1.Cohort, err error)
+	// Cohorts returns an object that can list and get Cohorts.
+	Cohorts(namespace string) CohortNamespaceLister
+	CohortListerExpansion
+}
+
+// cohortLister implements the CohortLister interface.
+type cohortLister struct {
+	listers.ResourceIndexer[*kueuev1alpha1.Cohort]
+}
+
+// NewCohortLister returns a new CohortLister.
+func NewCohortLister(indexer cache.Indexer) CohortLister {
+	return &cohortLister{listers.New[*kueuev1alpha1.Cohort](indexer, kueuev1alpha1.Resource("cohort"))}
+}
+
+// Cohorts returns an object that can list and get Cohorts.
+func (s *cohortLister) Cohorts(namespace string) CohortNamespaceLister {
+	return cohortNamespaceLister{listers.NewNamespaced[*kueuev1alpha1.Cohort](s.ResourceIndexer, namespace)}
+}
+
+// CohortNamespaceLister helps list and get Cohorts.
+// All objects returned here must be treated as read-only.
+type CohortNamespaceLister interface {
+	// List lists all Cohorts in the indexer for a given namespace.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*kueuev1alpha1.Cohort, err error)
+	// Get retrieves the Cohort from the indexer for a given namespace and name.
+	// Objects returned here must be treated as read-only.
+	Get(name string) (*kueuev1alpha1.Cohort, error)
+	CohortNamespaceListerExpansion
+}
+
+// cohortNamespaceLister implements the CohortNamespaceLister
+// interface.
+type cohortNamespaceLister struct {
+	listers.ResourceIndexer[*kueuev1alpha1.Cohort]
+}
diff --git a/client-go/listers/kueue/v1alpha1/expansion_generated.go b/client-go/listers/kueue/v1alpha1/expansion_generated.go
index a760283dd..a9cc1fd6d 100644
--- a/client-go/listers/kueue/v1alpha1/expansion_generated.go
+++ b/client-go/listers/kueue/v1alpha1/expansion_generated.go
@@ -17,6 +17,14 @@ limitations under the License.
 
 package v1alpha1
 
+// CohortListerExpansion allows custom methods to be added to
+// CohortLister.
+type CohortListerExpansion interface{}
+
+// CohortNamespaceListerExpansion allows custom methods to be added to
+// CohortNamespaceLister.
+type CohortNamespaceListerExpansion interface{}
+
 // TopologyListerExpansion allows custom methods to be added to
 // TopologyLister.
 type TopologyListerExpansion interface{}
diff --git a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
index 34e0f1dad..1a3b59538 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
@@ -8299,6 +8299,18 @@ spec:
                             - JobSet: kubernetes.io/job-completion-index (inherited from Job)
                             - Kubeflow: training.kubeflow.org/replica-index
                           type: string
+                        podSetSliceRequiredTopology:
+                          description: |-
+                            PodSetSliceRequiredTopology indicates the topology level required by the PodSet slice, as
+                            indicated by the `kueue.x-k8s.io/podset-slice-required-topology` annotation.
+                          type: string
+                        podSetSliceSize:
+                          description: |-
+                            PodSetSliceSize indicates the size of a subgroup of pods in a PodSet for which
+                            Kueue finds a requested topology domain on a level defined
+                            in `kueue.x-k8s.io/podset-slice-required-topology` annotation.
+                          format: int32
+                          type: integer
                         preferred:
                           description: |-
                             preferred indicates the topology level preferred by the PodSet, as
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index 0064ebe81..597522adc 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -790,6 +790,9 @@ func (m *Manager) DeleteSecondPassWithoutLock(w *kueue.Workload) {
 // delay.
 func (m *Manager) QueueSecondPassIfNeeded(ctx context.Context, w *kueue.Workload) bool {
 	if workload.NeedsSecondPass(w) {
+		log := ctrl.LoggerFrom(ctx)
+		log.V(3).Info("Workload pre-queued for second pass", "workload", workload.Key(w))
+		m.secondPassQueue.prequeue(w)
 		m.clock.AfterFunc(time.Second, func() {
 			m.queueSecondPass(ctx, w)
 		})
@@ -801,14 +804,11 @@ func (m *Manager) QueueSecondPassIfNeeded(ctx context.Context, w *kueue.Workload
 func (m *Manager) queueSecondPass(ctx context.Context, w *kueue.Workload) {
 	m.Lock()
 	defer m.Unlock()
-	m.queueSecondPassWithoutLock(ctx, w)
-}
 
-func (m *Manager) queueSecondPassWithoutLock(ctx context.Context, w *kueue.Workload) {
 	log := ctrl.LoggerFrom(ctx)
-	log.V(3).Info("Workload queued for second pass of scheduling", "workload", workload.Key(w))
-
 	wInfo := workload.NewInfo(w, m.workloadInfoOptions...)
-	m.secondPassQueue.offer(wInfo)
-	m.Broadcast()
+	if m.secondPassQueue.queue(wInfo) {
+		log.V(3).Info("Workload queued for second pass of scheduling", "workload", workload.Key(w))
+		m.Broadcast()
+	}
 }
diff --git a/pkg/queue/second_pass_queue.go b/pkg/queue/second_pass_queue.go
index 4492d9222..abb2dc71d 100644
--- a/pkg/queue/second_pass_queue.go
+++ b/pkg/queue/second_pass_queue.go
@@ -19,18 +19,23 @@ package queue
 import (
 	"sync"
 
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
 type secondPassQueue struct {
 	sync.RWMutex
 
-	state map[workload.WorkloadReference]*workload.Info
+	prequeued sets.Set[workload.WorkloadReference]
+	queued    map[workload.WorkloadReference]*workload.Info
 }
 
 func newSecondPassQueue() *secondPassQueue {
 	return &secondPassQueue{
-		state: make(map[workload.WorkloadReference]*workload.Info),
+		prequeued: sets.New[workload.WorkloadReference](),
+		queued:    make(map[workload.WorkloadReference]*workload.Info),
 	}
 }
 
@@ -39,23 +44,37 @@ func (q *secondPassQueue) takeAllReady() []workload.Info {
 	defer q.Unlock()
 
 	var result []workload.Info
-	for _, v := range q.state {
+	for _, v := range q.queued {
 		result = append(result, *v)
 	}
-	q.state = make(map[workload.WorkloadReference]*workload.Info)
+	q.queued = make(map[workload.WorkloadReference]*workload.Info)
 	return result
 }
 
-func (q *secondPassQueue) offer(w *workload.Info) {
+func (q *secondPassQueue) prequeue(obj *kueue.Workload) {
 	q.Lock()
 	defer q.Unlock()
 
-	q.state[workload.Key(w.Obj)] = w
+	q.prequeued.Insert(workload.Key(obj))
+}
+
+func (q *secondPassQueue) queue(w *workload.Info) bool {
+	q.Lock()
+	defer q.Unlock()
+
+	key := workload.Key(w.Obj)
+	enqueued := q.prequeued.Has(key) && workload.NeedsSecondPass(w.Obj)
+	if enqueued {
+		q.queued[key] = w
+	}
+	q.prequeued.Delete(key)
+	return enqueued
 }
 
 func (q *secondPassQueue) deleteByKey(key workload.WorkloadReference) {
 	q.Lock()
 	defer q.Unlock()
 
-	delete(q.state, key)
+	delete(q.queued, key)
+	q.prequeued.Delete(key)
 }
diff --git a/site/content/en/docs/reference/kueue-alpha.v1alpha1.md b/site/content/en/docs/reference/kueue-alpha.v1alpha1.md
index 620af7793..06318fec3 100644
--- a/site/content/en/docs/reference/kueue-alpha.v1alpha1.md
+++ b/site/content/en/docs/reference/kueue-alpha.v1alpha1.md
@@ -10,17 +10,21 @@ description: Generated API reference documentation for kueue.x-k8s.io/v1alpha1.
 ## Resource Types 
 
 
+- [Cohort](#kueue-x-k8s-io-v1alpha1-Cohort)
 - [Topology](#kueue-x-k8s-io-v1alpha1-Topology)
   
 
-## `Topology`     {#kueue-x-k8s-io-v1alpha1-Topology}
+## `Cohort`     {#kueue-x-k8s-io-v1alpha1-Cohort}
     
 
 **Appears in:**
 
 
 
-<p>Topology is the Schema for the topology API</p>
+<p>Cohort defines the Cohorts API.</p>
+<p>Hierarchical Cohorts (any Cohort which has a parent) are compatible
+with Fair Sharing as of v0.11. Using these features together in
+V0.9 and V0.10 is unsupported, and results in undefined behavior.</p>
 
 
 <table class="table">
@@ -28,11 +32,17 @@ description: Generated API reference documentation for kueue.x-k8s.io/v1alpha1.
 <tbody>
     
 <tr><td><code>apiVersion</code><br/>string</td><td><code>kueue.x-k8s.io/v1alpha1</code></td></tr>
-<tr><td><code>kind</code><br/>string</td><td><code>Topology</code></td></tr>
+<tr><td><code>kind</code><br/>string</td><td><code>Cohort</code></td></tr>
     
   
 <tr><td><code>spec</code> <B>[Required]</B><br/>
-<a href="#kueue-x-k8s-io-v1alpha1-TopologySpec"><code>TopologySpec</code></a>
+<a href="#kueue-x-k8s-io-v1alpha1-CohortSpec"><code>CohortSpec</code></a>
+</td>
+<td>
+   <span class="text-muted">No description provided.</span></td>
+</tr>
+<tr><td><code>status</code> <B>[Required]</B><br/>
+<a href="#kueue-x-k8s-io-v1alpha1-CohortStatus"><code>CohortStatus</code></a>
 </td>
 <td>
    <span class="text-muted">No description provided.</span></td>
@@ -40,32 +50,26 @@ description: Generated API reference documentation for kueue.x-k8s.io/v1alpha1.
 </tbody>
 </table>
 
-## `Cohort`     {#kueue-x-k8s-io-v1alpha1-Cohort}
+## `Topology`     {#kueue-x-k8s-io-v1alpha1-Topology}
     
 
 **Appears in:**
 
 
 
-<p>Cohort defines the Cohorts API.</p>
-<p>Hierarchical Cohorts (any Cohort which has a parent) are compatible
-with Fair Sharing as of v0.11. Using these features together in
-V0.9 and V0.10 is unsupported, and results in undefined behavior.</p>
+<p>Topology is the Schema for the topology API</p>
 
 
 <table class="table">
 <thead><tr><th width="30%">Field</th><th>Description</th></tr></thead>
 <tbody>
     
+<tr><td><code>apiVersion</code><br/>string</td><td><code>kueue.x-k8s.io/v1alpha1</code></td></tr>
+<tr><td><code>kind</code><br/>string</td><td><code>Topology</code></td></tr>
+    
   
 <tr><td><code>spec</code> <B>[Required]</B><br/>
-<a href="#kueue-x-k8s-io-v1alpha1-CohortSpec"><code>CohortSpec</code></a>
-</td>
-<td>
-   <span class="text-muted">No description provided.</span></td>
-</tr>
-<tr><td><code>status</code> <B>[Required]</B><br/>
-<a href="#kueue-x-k8s-io-v1alpha1-CohortStatus"><code>CohortStatus</code></a>
+<a href="#kueue-x-k8s-io-v1alpha1-TopologySpec"><code>TopologySpec</code></a>
 </td>
 <td>
    <span class="text-muted">No description provided.</span></td>
diff --git a/site/content/en/docs/reference/kueue.v1beta1.md b/site/content/en/docs/reference/kueue.v1beta1.md
index 504941ebe..2c7783fb3 100644
--- a/site/content/en/docs/reference/kueue.v1beta1.md
+++ b/site/content/en/docs/reference/kueue.v1beta1.md
@@ -2045,6 +2045,23 @@ within a PodSet. For example, in the context of JobSet this is jobset.sigs.k8s.i
 For example, in the context of JobSet this value is read from jobset.sigs.k8s.io/replicatedjob-replicas.</p>
 </td>
 </tr>
+<tr><td><code>podSetSliceRequiredTopology</code><br/>
+<code>string</code>
+</td>
+<td>
+   <p>PodSetSliceRequiredTopology indicates the topology level required by the PodSet slice, as
+indicated by the <code>kueue.x-k8s.io/podset-slice-required-topology</code> annotation.</p>
+</td>
+</tr>
+<tr><td><code>podSetSliceSize</code><br/>
+<code>int32</code>
+</td>
+<td>
+   <p>PodSetSliceSize indicates the size of a subgroup of pods in a PodSet for which
+Kueue finds a requested topology domain on a level defined
+in <code>kueue.x-k8s.io/podset-slice-required-topology</code> annotation.</p>
+</td>
+</tr>
 </tbody>
 </table>
 



Test patch:
diff --git a/pkg/queue/manager_test.go b/pkg/queue/manager_test.go
index 49f0a6bd5..05c2d21e5 100644
--- a/pkg/queue/manager_test.go
+++ b/pkg/queue/manager_test.go
@@ -26,9 +26,11 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
+	testingclock "k8s.io/utils/clock/testing"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
@@ -1285,3 +1287,87 @@ func TestGetPendingWorkloadsInfo(t *testing.T) {
 		})
 	}
 }
+
+func TestQueueSecondPassIfNeeded(t *testing.T) {
+	now := time.Now()
+
+	baseWorkloadBuilder := utiltesting.MakeWorkload("foo", "default").
+		Queue("tas-main").
+		PodSets(*utiltesting.MakePodSet("one", 1).
+			RequiredTopologyRequest(corev1.LabelHostname).
+			Request(corev1.ResourceCPU, "1").
+			Obj())
+	baseWorkloadNeedingSecondPass := baseWorkloadBuilder.Clone().
+		ReserveQuota(
+			utiltesting.MakeAdmission("tas-main", "one").
+				Assignment(corev1.ResourceCPU, "tas-default", "1000m").
+				DelayedTopologyRequest(kueue.DelayedTopologyRequestStatePending).
+				AssignmentPodCount(1).Obj(),
+		).
+		AdmissionCheck(kueue.AdmissionCheckState{
+			Name:  "prov-check",
+			State: kueue.CheckStateReady,
+		})
+	baseWorkloadNotNeedingSecondPass := baseWorkloadBuilder.Clone()
+
+	cases := map[string]struct {
+		workloads []*kueue.Workload
+		deleted   sets.Set[workload.WorkloadReference]
+		passTime  time.Duration
+		wantReady sets.Set[workload.WorkloadReference]
+	}{
+		"single queued workload checked immediately": {
+			workloads: []*kueue.Workload{baseWorkloadNeedingSecondPass.Obj()},
+		},
+		"single queued workload checked after 1s": {
+			workloads: []*kueue.Workload{
+				baseWorkloadNeedingSecondPass.DeepCopy(),
+				baseWorkloadNotNeedingSecondPass.DeepCopy(),
+			},
+			passTime:  time.Second,
+			wantReady: sets.New(workload.Key(baseWorkloadNeedingSecondPass.Obj())),
+		},
+		"single queued workload deleted in the meanwhile": {
+			workloads: []*kueue.Workload{
+				baseWorkloadNeedingSecondPass.DeepCopy(),
+				baseWorkloadNotNeedingSecondPass.DeepCopy(),
+			},
+			deleted:  sets.New(workload.Key(baseWorkloadNeedingSecondPass.Obj())),
+			passTime: time.Second,
+		},
+		"two queued workloads, one deleted in the meanwhile": {
+			workloads: []*kueue.Workload{
+				baseWorkloadNeedingSecondPass.Clone().Name("second").Obj(),
+			},
+			deleted:   sets.New(workload.Key(baseWorkloadNeedingSecondPass.Obj())),
+			passTime:  time.Second,
+			wantReady: sets.New(workload.NewWorkloadReference("default", "second")),
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			ctx := t.Context()
+
+			fakeClock := testingclock.NewFakeClock(now)
+			opts := []Option{
+				WithClock(fakeClock),
+			}
+			manager := NewManager(utiltesting.NewFakeClient(), nil, opts...)
+
+			for _, wl := range tc.workloads {
+				manager.QueueSecondPassIfNeeded(ctx, wl)
+			}
+			for _, wl := range tc.deleted.UnsortedList() {
+				manager.secondPassQueue.deleteByKey(wl)
+			}
+			fakeClock.Step(tc.passTime)
+			gotReady := sets.New[workload.WorkloadReference]()
+			for _, wlInfo := range manager.secondPassQueue.takeAllReady() {
+				gotReady.Insert(workload.Key(wlInfo.Obj))
+			}
+			if diff := cmp.Diff(tc.wantReady, gotReady); diff != "" {
+				t.Errorf("Unexpected ready workloads returned (-want,+got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index 9a100b26d..715ec804a 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -227,8 +227,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 			util.MustCreate(ctx, k8sClient, admissionCheck)
 			util.SetAdmissionCheckActive(ctx, k8sClient, admissionCheck, metav1.ConditionTrue)
 
-			clusterQueue = testing.MakeClusterQueue("").
-				GeneratedName("cluster-queue-").
+			clusterQueue = testing.MakeClusterQueue("cq").
 				ResourceGroup(
 					*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj(),
 				).AdmissionChecks(kueue.AdmissionCheckReference(admissionCheck.Name)).Obj()
@@ -319,8 +318,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					TopologyName("default").Obj()
 				util.MustCreate(ctx, k8sClient, tasFlavor)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
 					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj()).
 					Obj()
 				util.MustCreate(ctx, k8sClient, clusterQueue)
@@ -798,8 +796,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					TopologyName("default").Obj()
 				util.MustCreate(ctx, k8sClient, tasFlavor)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
 					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj()).
 					Obj()
 				util.MustCreate(ctx, k8sClient, clusterQueue)
@@ -1264,8 +1261,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					TopologyName("default").Obj()
 				util.MustCreate(ctx, k8sClient, tasFlavor)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
 					Preemption(kueue.ClusterQueuePreemption{
 						WithinClusterQueue: kueue.PreemptionPolicyLowerPriority,
 					}).
@@ -1388,8 +1384,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					TopologyName("default").Obj()
 				util.MustCreate(ctx, k8sClient, tasFlavor)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
 					Cohort("cohort").
 					Preemption(kueue.ClusterQueuePreemption{
 						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
@@ -1405,8 +1400,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 				localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
 				util.MustCreate(ctx, k8sClient, localQueue)
 
-				clusterQueueB = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueueB = testing.MakeClusterQueue("cluster-queue-b").
 					Cohort("cohort").
 					Preemption(kueue.ClusterQueuePreemption{
 						WithinClusterQueue:  kueue.PreemptionPolicyLowerPriority,
@@ -1493,8 +1487,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					Obj()
 				util.MustCreate(ctx, k8sClient, tasFlavor)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
 					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj()).
 					Obj()
 				util.MustCreate(ctx, k8sClient, clusterQueue)
@@ -1573,8 +1566,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					Obj()
 				util.MustCreate(ctx, k8sClient, tasFlavor)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
 					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj()).
 					Obj()
 				util.MustCreate(ctx, k8sClient, clusterQueue)
@@ -1759,8 +1751,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 				util.MustCreate(ctx, k8sClient, ac)
 				util.SetAdmissionCheckActive(ctx, k8sClient, ac, metav1.ConditionTrue)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cq").
 					ResourceGroup(
 						*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj(),
 					).AdmissionChecks(kueue.AdmissionCheckReference(ac.Name)).Obj()
@@ -2346,8 +2337,7 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 					Obj()
 				util.MustCreate(ctx, k8sClient, tasCPUFlavor)
 
-				clusterQueue = testing.MakeClusterQueue("").
-					GeneratedName("cluster-queue-").
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
 					ResourceGroup(
 						*testing.MakeFlavorQuotas(tasGPUFlavor.Name).Resource(corev1.ResourceCPU, "1").Resource(gpuResName, "5").Obj(),
 						*testing.MakeFlavorQuotas(tasCPUFlavor.Name).Resource(corev1.ResourceCPU, "5").Resource(gpuResName, "0").Obj(),