Repository: kubernetes-sigs/kueue

Issue:
Deleted workload entry outlives cq in scheduler cache
As we see in this [log](https://storage.googleapis.com/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/5461/pull-kueue-test-integration-baseline-main/1930264942032719872/build-log.txt) and many examples in the #5395.

Workload `wl1` entry from test `Nodes are created before test with the hostname being the lowest level  should evict workload when multiple assigned nodes fail` outlives the test in the scheduler's cache.
That's because it was re-queued for second pass.
Thus it is still visible in test `Preemption is enabled within ClusterQueue  should preempt the low and mid priority workloads to fit the high-priority workload` in result it is considered as possible preemption for `wl2` causing preemption failure
```
  2025-06-04T14:11:53.966698632Z	ERROR	scheduler	scheduler/scheduler.go:271	Failed to preempt workloads	{"schedulingCycle": 167, "workload": {"name":"wl2","namespace":"tas-rjx27"}, "clusterQueue": {"name":"cluster-queue"}, "error": "workloads.kueue.x-k8s.io \"wl1\" not found"}
```
and blocking `wl2` from being admitted.

**What you expected to happen**:
Delete workloads don't get scheduled for second pass, if they do scheduler do not nominates them for preemption.

**How to reproduce it (as minimally and precisely as possible)**:
Here is the reproduction [PR](https://github.com/kubernetes-sigs/kueue/pull/5513)


Discussion:
@mimowo: /assign 
tentatively, I hope to have time this week, at least to investigate and propose how to fix@mimowo: @mszadkow could you leave a reference to the repro test that you created? Maybe even leave it as a gist, or reference to your branch.@mszadkow: > [@mszadkow](https://github.com/mszadkow) could you leave a reference to the repro test that you created? Maybe even leave it as a gist, or reference to your branch.

Added to the issue description@mimowo: thx@mimowo: Ok, I have analyzed the issue and I think I understand it. I will describe soonish, and I'm creating follow up issues which I discovered on the way, and which made the investigation harder than it should be, for now: https://github.com/kubernetes-sigs/kueue/issues/5559 and https://github.com/kubernetes-sigs/kueue/issues/5558@mimowo: Other issues which interfere with this bug and make it more difficult:
- we trigger second pass for Evicted or Finished workloads https://github.com/kubernetes-sigs/kueue/issues/5561 (not source of the issue, but making it harder to reason about). The flake still sometimes surfaces
- we use two step eviction https://github.com/kubernetes-sigs/kueue/issues/5560 (not source of the issue, but makes it harder to reason about). The flake would not surface if we did that.
- when  Kueue scheduler fails to preempt (due to the target workload removed) gives up and not retries (I will open soon, still playing to better understand the circumstances around the code [here](https://github.com/kubernetes-sigs/kueue/blob/2b0c22bdeb9ea3d46842820f0d7b46be412a576e/pkg/scheduler/scheduler.go#L266-L274)) cc @gabesaba 
- When forgetWorkload happens in the 1s delay [here](https://github.com/kubernetes-sigs/kueue/blob/2b0c22bdeb9ea3d46842820f0d7b46be412a576e/pkg/queue/manager.go#L793C21-L793C31), then the second pass is rescheduled still (main issue)@mimowo: > * when  Kueue scheduler fails to preempt (due to the target workload removed) gives up and not retries (I will open soon, still playing to better understand the circumstances around the code [here](https://github.com/kubernetes-sigs/kueue/blob/2b0c22bdeb9ea3d46842820f0d7b46be412a576e/pkg/scheduler/scheduler.go#L266-L274)) cc [@gabesaba](https://github.com/gabesaba)

Here is the issue: https://github.com/kubernetes-sigs/kueue/issues/5590 along with the repro PR https://github.com/kubernetes-sigs/kueue/pull/5589

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/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/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/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>



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(),