Repository: kubernetes-sigs/kueue

Issue:
Flaky E2E test: MultiKueue when Creating a multikueue admission check Should run a RayCluster on worker if admitted
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

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


**What happened**:
End To End MultiKueue Suite: kindest/node:v1.32.3: [It] MultiKueue when Creating a multikueue admission check Should run a RayCluster on worker if admitted

```
{Timed out after 45.001s.
The function passed to Eventually failed at /home/prow/go/src/sigs.k8s.io/kueue/test/e2e/multikueue/e2e_test.go:828 with:
Expected
    <int32>: 0
to equal
    <int32>: 1 failed [FAILED] Timed out after 45.001s.
The function passed to Eventually failed at /home/prow/go/src/sigs.k8s.io/kueue/test/e2e/multikueue/e2e_test.go:828 with:
Expected
    <int32>: 0
to equal
    <int32>: 1
In [It] at: /home/prow/go/src/sigs.k8s.io/kueue/test/e2e/multikueue/e2e_test.go:830 @ 04/01/25 07:23:48.388
}
```

**What you expected to happen**:
No errors

**How to reproduce it (as minimally and precisely as possible)**:
https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/4846/pull-kueue-test-e2e-multikueue-main/1906965243951583232

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

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

Hints from issue comments:
@mbobrovskyi: /kind flake@mbobrovskyi: cc: @mszadkow @KPostOffice: The `Readiness` probe succeeds [here](https://storage.googleapis.com/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/4846/pull-kueue-test-e2e-multikueue-main/1906965243951583232/artifacts/run-test-multikueue-e2e-1.32.3/kind-worker1-control-plane/kubelet.log#:~:text=Apr%2001%2007%3A23%3A47%20kind%2Dworker1%2Dcontrol%2Dplane%20kubelet%5B775%5D%3A%20I0401%2007%3A23%3A47.744698%20%20%20%20%20775%20prober.go%3A116%5D%20%22Probe%20succeeded%22%20probeType%3D%22Readiness%22%20pod%3D%22multikueue%2D2db8t/raycluster1%2Dworkers%2Dgroup%2D0%2Dworker%2Dt94gd%22%20podUID%3D%221debb19f%2D6977%2D4287%2Da787%2Db9f3f2cf7ef2%22%20containerName%3D%22worker%2Dcontainer%22) in the Kubelet logs@mimowo: Would waiting a bit longer help? Maybe we have a case for increasing the CPU to 12 actually. I think we set 10 before introducing Ray and AppWrapper operators, but it would be good to see it we are CPU constrained by the dashboard@mbobrovskyi: It happens again https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/5013/pull-kueue-test-e2e-multikueue-main/1912548758457946112.@mszadkow: I found one more failed run here: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/4917/pull-kueue-test-e2e-multikueue-main/1912169469686321152

I will investigate
/assign@mszadkow: 1. That was with the `rayproject/ray:2.9.0`. 
Now it's `ray-mini`, which allows to get up the ray cluster worker pods much faster. 
2. I would still increase timeout here from `LongTimeout` to `VeryLongTimeout` as only this one doesn't have extended timeout like other test.
Which makes perfect sense as the full ray image is still be used in `periodic` tests.


Solution patch:
diff --git a/apis/kueue/v1alpha1/topology_types.go b/apis/kueue/v1alpha1/topology_types.go
index 84ba98f03..a24fb02d8 100644
--- a/apis/kueue/v1alpha1/topology_types.go
+++ b/apis/kueue/v1alpha1/topology_types.go
@@ -120,6 +120,7 @@ type Topology struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
 
+	// +kubebuilder:validation:Required
 	Spec TopologySpec `json:"spec,omitempty"`
 }
 
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml
index 0fb92bf84..edf0519c0 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_topologies.yaml
@@ -91,6 +91,8 @@ spec:
             required:
             - levels
             type: object
+        required:
+        - spec
         type: object
     served: true
     storage: true
diff --git a/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml b/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml
index a8e6bd866..14b6debc6 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_topologies.yaml
@@ -76,6 +76,8 @@ spec:
             required:
             - levels
             type: object
+        required:
+        - spec
         type: object
     served: true
     storage: true
diff --git a/go.mod b/go.mod
index c72614c98..f737253a5 100644
--- a/go.mod
+++ b/go.mod
@@ -20,6 +20,7 @@ require (
 	github.com/spf13/cobra v1.9.1
 	github.com/spf13/pflag v1.0.6
 	go.uber.org/zap v1.27.0
+	golang.org/x/sync v0.12.0
 	k8s.io/api v0.32.3
 	k8s.io/apimachinery v0.32.3
 	k8s.io/apiserver v0.32.3
@@ -125,7 +126,6 @@ require (
 	golang.org/x/mod v0.24.0 // indirect
 	golang.org/x/net v0.37.0 // indirect
 	golang.org/x/oauth2 v0.28.0 // indirect
-	golang.org/x/sync v0.12.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect
 	golang.org/x/term v0.30.0 // indirect
 	golang.org/x/text v0.23.0 // indirect
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
index d39814fef..29d89c12b 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
@@ -19,8 +19,11 @@ package leaderworkerset
 import (
 	"context"
 	"fmt"
+	"maps"
+	"slices"
 
 	"github.com/go-logr/logr"
+	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/tools/record"
@@ -28,15 +31,19 @@ import (
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
 	"sigs.k8s.io/kueue/pkg/features"
+	"sigs.k8s.io/kueue/pkg/util/parallelize"
+	utilslices "sigs.k8s.io/kueue/pkg/util/slices"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
@@ -92,7 +99,30 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	log := ctrl.LoggerFrom(ctx)
 	log.V(2).Info("Reconcile LeaderWorkerSet")
 
-	err = r.createPrebuiltWorkloadsIfNotExist(ctx, lws)
+	wlList := &kueue.WorkloadList{}
+	if err := r.client.List(ctx, wlList, client.InNamespace(lws.GetNamespace()),
+		client.MatchingFields{indexer.OwnerReferenceUID: string(lws.GetUID())},
+	); err != nil {
+		return ctrl.Result{}, err
+	}
+
+	toCreate, toFinalize := r.filterWorkloads(lws, wlList.Items)
+
+	eg, ctx := errgroup.WithContext(ctx)
+
+	eg.Go(func() error {
+		return parallelize.Until(ctx, len(toCreate), func(i int) error {
+			return r.createPrebuiltWorkload(ctx, lws, toCreate[i])
+		})
+	})
+
+	eg.Go(func() error {
+		return parallelize.Until(ctx, len(toFinalize), func(i int) error {
+			return r.removeOwnerReference(ctx, lws, toFinalize[i])
+		})
+	})
+
+	err = eg.Wait()
 	if err != nil {
 		return ctrl.Result{}, err
 	}
@@ -100,28 +130,38 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	return ctrl.Result{}, nil
 }
 
-func (r *Reconciler) createPrebuiltWorkloadsIfNotExist(ctx context.Context, lws *leaderworkersetv1.LeaderWorkerSet) error {
-	replicas := ptr.Deref(lws.Spec.Replicas, 1)
+// filterWorkloads compares the desired state in a LeaderWorkerSet with existing workloads,
+// identifying workloads to create and those to finalize.
+//
+// It takes a LeaderWorkerSet and a slice of existing Workload objects as input and returns:
+// 1. A slice of workload names that need to be created
+// 2. A slice of Workload pointers that need to be finalized
+func (r *Reconciler) filterWorkloads(lws *leaderworkersetv1.LeaderWorkerSet, existingWorkloads []kueue.Workload) ([]string, []*kueue.Workload) {
+	var (
+		toCreate   []string
+		toFinalize = utilslices.ToRefMap(existingWorkloads, func(e *kueue.Workload) string {
+			return e.Name
+		})
+		replicas = ptr.Deref(lws.Spec.Replicas, 1)
+	)
+
 	for i := int32(0); i < replicas; i++ {
-		if err := r.createPrebuiltWorkloadIfNotExist(ctx, lws, i); err != nil {
-			return err
+		workloadName := GetWorkloadName(lws.UID, lws.Name, fmt.Sprint(i))
+		if _, ok := toFinalize[workloadName]; ok {
+			delete(toFinalize, workloadName)
+		} else {
+			toCreate = append(toCreate, workloadName)
 		}
 	}
-	return nil
+
+	return toCreate, slices.Collect(maps.Values(toFinalize))
 }
 
-func (r *Reconciler) createPrebuiltWorkloadIfNotExist(ctx context.Context, lws *leaderworkersetv1.LeaderWorkerSet, index int32) error {
-	wl := &kueue.Workload{}
-	err := r.client.Get(ctx, client.ObjectKey{Name: GetWorkloadName(lws.UID, lws.Name, fmt.Sprint(index)), Namespace: lws.Namespace}, wl)
-	// Ignore if the Workload already exists or an error occurs.
-	if err == nil || client.IgnoreNotFound(err) != nil {
+func (r *Reconciler) createPrebuiltWorkload(ctx context.Context, lws *leaderworkersetv1.LeaderWorkerSet, workloadName string) error {
+	createdWorkload, err := r.constructWorkload(lws, workloadName)
+	if err != nil {
 		return err
 	}
-	return r.createPrebuiltWorkload(ctx, lws, index)
-}
-
-func (r *Reconciler) createPrebuiltWorkload(ctx context.Context, lws *leaderworkersetv1.LeaderWorkerSet, index int32) error {
-	createdWorkload := r.constructWorkload(lws, index)
 
 	priorityClassName, source, p, err := jobframework.ExtractPriority(ctx, r.client, lws, createdWorkload.Spec.PodSets, nil)
 	if err != nil {
@@ -143,8 +183,12 @@ func (r *Reconciler) createPrebuiltWorkload(ctx context.Context, lws *leaderwork
 	return nil
 }
 
-func (r *Reconciler) constructWorkload(lws *leaderworkersetv1.LeaderWorkerSet, index int32) *kueue.Workload {
-	return podcontroller.NewGroupWorkload(GetWorkloadName(lws.UID, lws.Name, fmt.Sprint(index)), lws, r.podSets(lws), r.labelKeysToCopy)
+func (r *Reconciler) constructWorkload(lws *leaderworkersetv1.LeaderWorkerSet, workloadName string) (*kueue.Workload, error) {
+	createdWorkload := podcontroller.NewGroupWorkload(workloadName, lws, r.podSets(lws), r.labelKeysToCopy)
+	if err := controllerutil.SetOwnerReference(lws, createdWorkload, r.client.Scheme()); err != nil {
+		return nil, err
+	}
+	return createdWorkload, nil
 }
 
 func (r *Reconciler) podSets(lws *leaderworkersetv1.LeaderWorkerSet) []kueue.PodSet {
@@ -198,6 +242,14 @@ func (r *Reconciler) podSets(lws *leaderworkersetv1.LeaderWorkerSet) []kueue.Pod
 	return podSets
 }
 
+func (r *Reconciler) removeOwnerReference(ctx context.Context, lws *leaderworkersetv1.LeaderWorkerSet, wl *kueue.Workload) error {
+	err := controllerutil.RemoveOwnerReference(lws, wl, r.client.Scheme())
+	if err != nil {
+		return err
+	}
+	return r.client.Update(ctx, wl)
+}
+
 var _ predicate.Predicate = (*Reconciler)(nil)
 
 func (r *Reconciler) Generic(event.GenericEvent) bool {
diff --git a/pkg/controller/jobs/pod/pod_controller.go b/pkg/controller/jobs/pod/pod_controller.go
index 34c25c7b4..846f87f14 100644
--- a/pkg/controller/jobs/pod/pod_controller.go
+++ b/pkg/controller/jobs/pod/pod_controller.go
@@ -340,6 +340,10 @@ func (p *Pod) RestorePodSetsInfo(_ []podset.PodSetInfo) bool {
 // Finished means whether the job is completed/failed or not,
 // condition represents the workload finished condition.
 func (p *Pod) Finished() (message string, success, finished bool) {
+	if p.isServing() {
+		return "", true, false
+	}
+
 	finished = true
 	success = true
 
diff --git a/site/content/en/docs/concepts/multikueue.md b/site/content/en/docs/concepts/multikueue.md
index b41e505f3..f2c198434 100644
--- a/site/content/en/docs/concepts/multikueue.md
+++ b/site/content/en/docs/concepts/multikueue.md
@@ -78,7 +78,7 @@ On the other hand, the Worker cluster should install the full kubeflow operator.
 
 ## Plain Pods
 
-MultiKueue supports the remote creation and management of Plain Pods.
+MultiKueue supports the remote creation and management of Plain Pods and Group of Pods.
 
 ## Deployments
 
diff --git a/site/content/en/docs/tasks/run/multikueue/plain_pods.md b/site/content/en/docs/tasks/run/multikueue/plain_pods.md
index becdde494..ea1f93f28 100644
--- a/site/content/en/docs/tasks/run/multikueue/plain_pods.md
+++ b/site/content/en/docs/tasks/run/multikueue/plain_pods.md
@@ -20,6 +20,10 @@ Pods created on the manager cluster are automatically gated and receive live sta
 
 ## Example
 
-Once the setup is complete you can test it by running the example below:
+Once the setup is complete you can test it by running the examples below:
 
+1. Single plain pod
 {{< include "examples/pods-kueue/kueue-pod.yaml" "yaml" >}}
+
+2. Group of pods
+{{< include "examples/pods-kueue/kueue-pod-group.yaml" "yaml" >}}
\ No newline at end of file


Test patch:
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
index 05a5dc358..2a0e64700 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
@@ -71,6 +71,7 @@ func TestReconciler(t *testing.T) {
 			wantLeaderWorkerSet: leaderworkerset.MakeLeaderWorkerSet(testLWS, testNS).UID(testUID).Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
@@ -128,6 +129,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
@@ -200,6 +202,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
@@ -231,6 +234,7 @@ func TestReconciler(t *testing.T) {
 					Priority(0).
 					Obj(),
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "1"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
@@ -345,6 +349,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
@@ -454,6 +459,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
@@ -511,6 +517,7 @@ func TestReconciler(t *testing.T) {
 				Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(
@@ -545,6 +552,89 @@ func TestReconciler(t *testing.T) {
 			},
 			enableTopologyAwareScheduling: false,
 		},
+		"should delete LeaderWorkerSet ownerReference from the redundant prebuilt workload": {
+			leaderWorkerSet:     leaderworkerset.MakeLeaderWorkerSet(testLWS, testNS).UID(testUID).Obj(),
+			wantLeaderWorkerSet: leaderworkerset.MakeLeaderWorkerSet(testLWS, testNS).UID(testUID).Obj(),
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
+					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "test-pod1", "test-pod1-uid").
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
+					Finalizers(kueue.ResourceInUseFinalizerName).
+					PodSets(
+						kueue.PodSet{
+							Name: kueue.DefaultPodSetName,
+							Template: corev1.PodTemplateSpec{
+								Spec: corev1.PodSpec{
+									Containers: []corev1.Container{
+										{Name: "c", Image: "pause"},
+									},
+								},
+							},
+							Count: 1,
+						}).
+					Priority(0).
+					Obj(),
+				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "1"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
+					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "test-pod2", "test-pod2-uid").
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
+					Finalizers(kueue.ResourceInUseFinalizerName).
+					PodSets(
+						kueue.PodSet{
+							Name: kueue.DefaultPodSetName,
+							Template: corev1.PodTemplateSpec{
+								Spec: corev1.PodSpec{
+									Containers: []corev1.Container{
+										{Name: "c", Image: "pause"},
+									},
+								},
+							},
+							Count: 1,
+						}).
+					Priority(0).
+					Obj(),
+			},
+			wantWorkloads: []kueue.Workload{
+				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					OwnerReference(gvk, testLWS, testUID).
+					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "test-pod1", "test-pod1-uid").
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
+					Finalizers(kueue.ResourceInUseFinalizerName).
+					PodSets(
+						kueue.PodSet{
+							Name: kueue.DefaultPodSetName,
+							Template: corev1.PodTemplateSpec{
+								Spec: corev1.PodSpec{
+									Containers: []corev1.Container{
+										{Name: "c", Image: "pause"},
+									},
+								},
+							},
+							Count: 1,
+						}).
+					Priority(0).
+					Obj(),
+				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "1"), testNS).
+					OwnerReference(corev1.SchemeGroupVersion.WithKind("Pod"), "test-pod2", "test-pod2-uid").
+					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
+					Finalizers(kueue.ResourceInUseFinalizerName).
+					PodSets(
+						kueue.PodSet{
+							Name: kueue.DefaultPodSetName,
+							Template: corev1.PodTemplateSpec{
+								Spec: corev1.PodSpec{
+									Containers: []corev1.Container{
+										{Name: "c", Image: "pause"},
+									},
+								},
+							},
+							Count: 1,
+						}).
+					Priority(0).
+					Obj(),
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
diff --git a/test/e2e/customconfigs/managejobswithoutqueuename_test.go b/test/e2e/customconfigs/managejobswithoutqueuename_test.go
index 502d92915..fdb69b04c 100644
--- a/test/e2e/customconfigs/managejobswithoutqueuename_test.go
+++ b/test/e2e/customconfigs/managejobswithoutqueuename_test.go
@@ -27,6 +27,7 @@ import (
 	"k8s.io/utils/ptr"
 	"k8s.io/utils/strings/slices"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/jobset/api/jobset/v1alpha2"
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
@@ -650,6 +651,9 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 					g.Expect(createdWorkloads.Items).To(gomega.HaveLen(1))
 					g.Expect(createdWorkloads.Items[0].Name).To(gomega.Equal(leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0")))
 					g.Expect(createdWorkloads.Items[0].Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+					hasOwnerRef, err := controllerutil.HasOwnerReference(createdWorkloads.Items[0].OwnerReferences, lws, k8sClient.Scheme())
+					g.Expect(err).ToNot(gomega.HaveOccurred())
+					g.Expect(hasOwnerRef).To(gomega.BeTrue())
 				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -673,6 +677,21 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 						g.Expect(pod.Status.Phase).Should(gomega.Equal(corev1.PodRunning))
 					}
 				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+
+				ginkgo.By("Delete the LeaderWorkerSet", func() {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
+				})
+
+				ginkgo.By("Check workload is deleted", func() {
+					wlLookupKey := types.NamespacedName{
+						Name:      leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"),
+						Namespace: ns.Name,
+					}
+					createdWorkload := &kueue.Workload{}
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(testing.BeNotFoundError())
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
 			})
 		})
 
@@ -695,6 +714,9 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 					g.Expect(createdWorkloads.Items).To(gomega.HaveLen(1))
 					g.Expect(createdWorkloads.Items[0].Name).To(gomega.Equal(leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0")))
 					g.Expect(createdWorkloads.Items[0].Status.Conditions).To(testing.HaveConditionStatusFalse(kueue.WorkloadQuotaReserved))
+					hasOwnerRef, err := controllerutil.HasOwnerReference(createdWorkloads.Items[0].OwnerReferences, lws, k8sClient.Scheme())
+					g.Expect(err).ToNot(gomega.HaveOccurred())
+					g.Expect(hasOwnerRef).To(gomega.BeTrue())
 				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -719,6 +741,21 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 					}
 				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
 			})
+
+			ginkgo.By("Delete the LeaderWorkerSet", func() {
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
+			})
+
+			ginkgo.By("Check workload is deleted", func() {
+				wlLookupKey := types.NamespacedName{
+					Name:      leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"),
+					Namespace: ns.Name,
+				}
+				createdWorkload := &kueue.Workload{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(testing.BeNotFoundError())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
 		})
 	})
 })
diff --git a/test/e2e/singlecluster/leaderworkerset_test.go b/test/e2e/singlecluster/leaderworkerset_test.go
index dc02ade9c..866fe2d7d 100644
--- a/test/e2e/singlecluster/leaderworkerset_test.go
+++ b/test/e2e/singlecluster/leaderworkerset_test.go
@@ -30,6 +30,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/leaderworkerset"
+	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	leaderworkersettesting "sigs.k8s.io/kueue/pkg/util/testingjobs/leaderworkerset"
 	"sigs.k8s.io/kueue/test/util"
@@ -449,6 +450,109 @@ var _ = ginkgo.Describe("LeaderWorkerSet integration", func() {
 			ginkgo.Entry("LeaderReadyStartupPolicy", leaderworkersetv1.LeaderReadyStartupPolicy),
 		)
 
+		ginkgo.DescribeTable("should allow to scale up, scale down fast",
+			func(startupPolicyType leaderworkersetv1.StartupPolicyType) {
+				lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
+					Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+					Size(3).
+					Replicas(1).
+					RequestAndLimit(corev1.ResourceCPU, "200m").
+					TerminationGracePeriod(1).
+					Queue(lq.Name).
+					StartupPolicy(startupPolicyType).
+					Obj()
+
+				ginkgo.By("Create a LeaderWorkerSet", func() {
+					gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
+				})
+
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
+
+				ginkgo.By("Waiting for replicas is ready", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+						g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+							gomega.BeComparableTo(metav1.Condition{
+								Type:   "Available",
+								Status: metav1.ConditionTrue,
+								Reason: "AllGroupsReady",
+							}, util.IgnoreConditionTimestampsAndObservedGeneration, util.IgnoreConditionMessage)),
+						)
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				createdWorkload1 := &kueue.Workload{}
+				wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
+				ginkgo.By("Check workload is created", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
+				})
+
+				ginkgo.By("Scale up LeaderWorkerSet", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						createdLeaderWorkerSet.Spec.Replicas = ptr.To[int32](2)
+						g.Expect(k8sClient.Update(ctx, createdLeaderWorkerSet)).To(gomega.Succeed())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Scale down LeaderWorkerSet", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						createdLeaderWorkerSet.Spec.Replicas = ptr.To[int32](1)
+						g.Expect(k8sClient.Update(ctx, createdLeaderWorkerSet)).To(gomega.Succeed())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Waiting for replicas is ready", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+						g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+							gomega.BeComparableTo(metav1.Condition{
+								Type:   "Available",
+								Status: metav1.ConditionTrue,
+								Reason: "AllGroupsReady",
+							}, util.IgnoreConditionTimestampsAndObservedGeneration, util.IgnoreConditionMessage)),
+						)
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workload for group 1 is still exist", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
+				})
+
+				createdWorkload2 := &kueue.Workload{}
+				wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "1"), Namespace: ns.Name}
+				ginkgo.By("Check workload for group 2 is deleted", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(testing.BeNotFoundError())
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Delete the LeaderWorkerSet", func() {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
+				})
+
+				ginkgo.By("Check pods are deleted", func() {
+					pods := &corev1.PodList{}
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+							leaderworkersetv1.SetNameLabelKey: lws.Name,
+						}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+						g.Expect(pods.Items).To(gomega.BeEmpty())
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workloads are deleted", func() {
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload2, false, util.LongTimeout)
+				})
+			},
+			ginkgo.Entry("LeaderCreatedStartupPolicy", leaderworkersetv1.LeaderCreatedStartupPolicy),
+			ginkgo.Entry("LeaderReadyStartupPolicy", leaderworkersetv1.LeaderReadyStartupPolicy),
+		)
+
 		ginkgo.DescribeTable("should admit group with multiple leaders and workers that fits and have different resource needs",
 			func(startupPolicyType leaderworkersetv1.StartupPolicyType) {
 				lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
@@ -789,6 +893,155 @@ var _ = ginkgo.Describe("LeaderWorkerSet integration", func() {
 					g.Expect(createdHighPriorityWl.Spec.PriorityClassName).To(gomega.Equal(highPriorityLWS.Name))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
+
+			ginkgo.By("Check the low priority Workload is preempted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, lowPriorityWlLookupKey, createdLowPriorityWl)).To(gomega.Succeed())
+					g.Expect(createdLowPriorityWl.Status.Conditions).To(testing.HaveConditionStatusFalse(kueue.WorkloadAdmitted))
+					g.Expect(createdLowPriorityWl.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadPreempted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+
+	ginkgo.When("Workload deactivated", func() {
+		ginkgo.It("shouldn't delete deactivated Workload", func() {
+			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
+				Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+				Size(3).
+				Replicas(1).
+				RequestAndLimit(corev1.ResourceCPU, "200m").
+				TerminationGracePeriod(1).
+				Queue(lq.Name).
+				Obj()
+
+			ginkgo.By("Create a LeaderWorkerSet", func() {
+				gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
+			})
+
+			ginkgo.By("Waiting for replicas is ready", func() {
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+						gomega.BeComparableTo(metav1.Condition{
+							Type:   "Available",
+							Status: metav1.ConditionTrue,
+							Reason: "AllGroupsReady",
+						}, util.IgnoreConditionTimestampsAndObservedGeneration, util.IgnoreConditionMessage)),
+					)
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			wlLookupKey := types.NamespacedName{
+				Name:      leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"),
+				Namespace: ns.Name,
+			}
+			createdWorkload := &kueue.Workload{}
+			ginkgo.By("Check workload is created", func() {
+				gomega.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+			})
+
+			createdWorkloadUID := createdWorkload.UID
+
+			ginkgo.By("Deactivate workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					createdWorkload.Spec.Active = ptr.To(false)
+					g.Expect(k8sClient.Update(ctx, createdWorkload)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Waiting for replicas to not be ready", func() {
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(0)))
+					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+						gomega.BeComparableTo(metav1.Condition{
+							Type:   "Available",
+							Status: metav1.ConditionFalse,
+							Reason: "AllGroupsReady",
+						}, util.IgnoreConditionTimestampsAndObservedGeneration, util.IgnoreConditionMessage)),
+					)
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Check pods are gated", func() {
+				pods := &corev1.PodList{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+						leaderworkersetv1.SetNameLabelKey: lws.Name,
+					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+					g.Expect(pods.Items).To(gomega.HaveLen(3))
+					for _, pod := range pods.Items {
+						g.Expect(pod.Spec.SchedulingGates).To(gomega.ContainElement(corev1.PodSchedulingGate{
+							Name: podconstants.SchedulingGateName,
+						}))
+					}
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Check workload is deactivated", func() {
+				gomega.Consistently(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					g.Expect(createdWorkload.UID).Should(gomega.Equal(createdWorkloadUID))
+					g.Expect(createdWorkload.Spec.Active).Should(gomega.Equal(ptr.To(false)))
+				}, util.ConsistentDuration, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Activate workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					createdWorkload.Spec.Active = ptr.To(true)
+					g.Expect(k8sClient.Update(ctx, createdWorkload)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Waiting for replicas to be ready", func() {
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
+
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+						gomega.BeComparableTo(metav1.Condition{
+							Type:   "Available",
+							Status: metav1.ConditionTrue,
+							Reason: "AllGroupsReady",
+						}, util.IgnoreConditionTimestampsAndObservedGeneration, util.IgnoreConditionMessage)),
+					)
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Check workload is admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					g.Expect(createdWorkload.UID).Should(gomega.Equal(createdWorkloadUID))
+					g.Expect(createdWorkload.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Delete the LeaderWorkerSet", func() {
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
+			})
+
+			ginkgo.By("Check pods are deleted", func() {
+				pods := &corev1.PodList{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+						leaderworkersetv1.SetNameLabelKey: lws.Name,
+					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+					g.Expect(pods.Items).To(gomega.BeEmpty())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Check workload is deleted", func() {
+				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload, false, util.LongTimeout)
+			})
 		})
 	})
 })
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index a62b83e11..3204682b0 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -1459,6 +1459,9 @@ var _ = ginkgo.Describe("Topology validations", func() {
 			ginkgo.Entry("valid Topology",
 				testing.MakeDefaultOneLevelTopology("valid"),
 				gomega.Succeed()),
+			ginkgo.Entry("no levels",
+				testing.MakeTopology("no-levels").Obj(),
+				testing.BeInvalidError()),
 			ginkgo.Entry("invalid levels",
 				testing.MakeTopology("invalid-level").Levels("@invalid").Obj(),
 				testing.BeInvalidError()),