Repository: kubernetes-sigs/kueue

Issue:
LWS workloads get deleted on preemption
**What happened**:

When LWS workload for a group is preempted it is deleted and recreated. See discussion in https://github.com/kubernetes-sigs/kueue/pull/4711/files#diff-8be708e6a6046713666b607889f5a2fd5f69f817a0a2815c62b03f725f6ab34fR798

However, this is an issue because then the state of the workload is lost. There might be important state with waitForPodsReady information. It also means the "active" field does not work.

**What you expected to happen**:

The preempted workload should not be deleted and recreated.

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

Follow the steps in the e2e test in https://github.com/kubernetes-sigs/kueue/pull/4711/files#diff-8be708e6a6046713666b607889f5a2fd5f69f817a0a2815c62b03f725f6ab34fR670

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

We have analogous issue for StatefulSet https://github.com/kubernetes-sigs/kueue/issues/4342

I think just adding owner reference for the workload to LWS would solve the problem.

Hints from issue comments:
@mimowo: /cc @mbobrovskyi 
@mbobrovskyi: /assign

Solution patch:
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index 344f6b560..ec01e22f7 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -1051,15 +1051,21 @@ func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl
 }
 
 func (r *JobReconciler) extractPriority(ctx context.Context, podSets []kueue.PodSet, job GenericJob) (string, string, int32, error) {
-	if workloadPriorityClass := WorkloadPriorityClassName(job.Object()); len(workloadPriorityClass) > 0 {
-		return utilpriority.GetPriorityFromWorkloadPriorityClass(ctx, r.client, workloadPriorityClass)
-	}
+	var customPriorityFunc func() string
 	if jobWithPriorityClass, isImplemented := job.(JobWithPriorityClass); isImplemented {
-		return utilpriority.GetPriorityFromPriorityClass(
-			ctx, r.client, jobWithPriorityClass.PriorityClass())
+		customPriorityFunc = jobWithPriorityClass.PriorityClass
+	}
+	return ExtractPriority(ctx, r.client, job.Object(), podSets, customPriorityFunc)
+}
+
+func ExtractPriority(ctx context.Context, c client.Client, obj client.Object, podSets []kueue.PodSet, customPriorityFunc func() string) (string, string, int32, error) {
+	if workloadPriorityClass := WorkloadPriorityClassName(obj); len(workloadPriorityClass) > 0 {
+		return utilpriority.GetPriorityFromWorkloadPriorityClass(ctx, c, workloadPriorityClass)
+	}
+	if customPriorityFunc != nil {
+		return utilpriority.GetPriorityFromPriorityClass(ctx, c, customPriorityFunc())
 	}
-	return utilpriority.GetPriorityFromPriorityClass(
-		ctx, r.client, extractPriorityFromPodSets(podSets))
+	return utilpriority.GetPriorityFromPriorityClass(ctx, c, extractPriorityFromPodSets(podSets))
 }
 
 func extractPriorityFromPodSets(podSets []kueue.PodSet) string {
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
index b171f96f1..907b74c68 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
@@ -127,10 +127,6 @@ func (r *PodReconciler) setDefault(ctx context.Context, pod *corev1.Pod) (bool,
 	pod.Labels[constants.QueueLabel] = queueName
 	pod.Labels[podconstants.GroupNameLabel] = wlName
 	pod.Labels[constants.PrebuiltWorkloadLabel] = wlName
-	if priorityClass := jobframework.WorkloadPriorityClassName(lws); priorityClass != "" {
-		pod.Labels[constants.WorkloadPriorityClassLabel] = priorityClass
-	}
-
 	pod.Annotations[podconstants.GroupTotalCountAnnotation] = fmt.Sprint(ptr.Deref(lws.Spec.LeaderWorkerTemplate.Size, 1))
 	pod.Annotations[podconstants.RoleHashAnnotation] = string(kueue.DefaultPodSetName)
 
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
index 9bf1e0d47..e49168e5c 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
@@ -112,7 +112,17 @@ func (r *Reconciler) createPrebuiltWorkloadIfNotExist(ctx context.Context, lws *
 
 func (r *Reconciler) createPrebuiltWorkload(ctx context.Context, lws *leaderworkersetv1.LeaderWorkerSet, index int32) error {
 	createdWorkload := r.constructWorkload(lws, index)
-	err := r.client.Create(ctx, createdWorkload)
+
+	priorityClassName, source, p, err := jobframework.ExtractPriority(ctx, r.client, lws, createdWorkload.Spec.PodSets, nil)
+	if err != nil {
+		return err
+	}
+
+	createdWorkload.Spec.PriorityClassName = priorityClassName
+	createdWorkload.Spec.Priority = &p
+	createdWorkload.Spec.PriorityClassSource = source
+
+	err = r.client.Create(ctx, createdWorkload)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
index af254b5f3..0c3a2a0ea 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
@@ -72,17 +72,24 @@ func (wh *Webhook) Default(ctx context.Context, obj runtime.Object) error {
 	}
 	if suspend {
 		if lws.Spec.LeaderWorkerTemplate.LeaderTemplate != nil {
-			wh.podTemplateSpecDefault(lws.Spec.LeaderWorkerTemplate.LeaderTemplate)
+			wh.podTemplateSpecDefault(lws, lws.Spec.LeaderWorkerTemplate.LeaderTemplate)
 		}
-		wh.podTemplateSpecDefault(&lws.Spec.LeaderWorkerTemplate.WorkerTemplate)
+		wh.podTemplateSpecDefault(lws, &lws.Spec.LeaderWorkerTemplate.WorkerTemplate)
 	}
 
 	return nil
 }
 
-func (wh *Webhook) podTemplateSpecDefault(podTemplateSpec *corev1.PodTemplateSpec) {
+func (wh *Webhook) podTemplateSpecDefault(lws *LeaderWorkerSet, podTemplateSpec *corev1.PodTemplateSpec) {
+	if priorityClass := jobframework.WorkloadPriorityClassName(lws.Object()); priorityClass != "" {
+		if podTemplateSpec.Labels == nil {
+			podTemplateSpec.Labels = make(map[string]string, 1)
+		}
+		podTemplateSpec.Labels[constants.WorkloadPriorityClassLabel] = priorityClass
+	}
+
 	if podTemplateSpec.Annotations == nil {
-		podTemplateSpec.Annotations = make(map[string]string, 1)
+		podTemplateSpec.Annotations = make(map[string]string, 2)
 	}
 	podTemplateSpec.Annotations[podconstants.SuspendedByParentAnnotation] = FrameworkName
 	podTemplateSpec.Annotations[podconstants.GroupServingAnnotationKey] = podconstants.GroupServingAnnotationValue



Test patch:
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
index 510658f3e..8f95a6409 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
@@ -28,7 +28,6 @@ import (
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
-	"sigs.k8s.io/kueue/pkg/controller/constants"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/util/testingjobs/leaderworkerset"
@@ -118,31 +117,6 @@ func TestPodReconciler(t *testing.T) {
 				Annotation(podconstants.RoleHashAnnotation, string(kueue.DefaultPodSetName)).
 				Obj(),
 		},
-		"should set default values and priority class when has value": {
-			lws: leaderworkerset.MakeLeaderWorkerSet("lws", "ns").
-				UID(testUID).
-				Queue("queue").
-				Label(constants.WorkloadPriorityClassLabel, "test").
-				Obj(),
-			pod: testingjobspod.MakePod("pod", "ns").
-				Label(leaderworkersetv1.SetNameLabelKey, "lws").
-				Label(leaderworkersetv1.GroupIndexLabelKey, "0").
-				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
-				Obj(),
-			wantPod: testingjobspod.MakePod("pod", "ns").
-				Label(leaderworkersetv1.SetNameLabelKey, "lws").
-				Label(leaderworkersetv1.GroupIndexLabelKey, "0").
-				Label(constants.WorkloadPriorityClassLabel, "test").
-				Queue("queue").
-				Group(GetWorkloadName(types.UID(testUID), "lws", "0")).
-				GroupTotalCount("1").
-				PrebuiltWorkload(GetWorkloadName(types.UID(testUID), "lws", "0")).
-				Annotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
-				Annotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
-				Annotation(podconstants.RoleHashAnnotation, string(kueue.DefaultPodSetName)).
-				Obj(),
-		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
index f1a99b694..ef9b7ecd1 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
@@ -33,6 +33,7 @@ import (
 
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -54,13 +55,14 @@ var (
 
 func TestReconciler(t *testing.T) {
 	cases := map[string]struct {
-		labelKeysToCopy     []string
-		leaderWorkerSet     *leaderworkersetv1.LeaderWorkerSet
-		workloads           []kueue.Workload
-		wantLeaderWorkerSet *leaderworkersetv1.LeaderWorkerSet
-		wantWorkloads       []kueue.Workload
-		wantEvents          []utiltesting.EventRecord
-		wantErr             error
+		labelKeysToCopy         []string
+		leaderWorkerSet         *leaderworkersetv1.LeaderWorkerSet
+		workloads               []kueue.Workload
+		workloadPriorityClasses []kueue.WorkloadPriorityClass
+		wantLeaderWorkerSet     *leaderworkersetv1.LeaderWorkerSet
+		wantWorkloads           []kueue.Workload
+		wantEvents              []utiltesting.EventRecord
+		wantErr                 error
 	}{
 		"should create prebuilt workload": {
 			leaderWorkerSet:     leaderworkerset.MakeLeaderWorkerSet(testLWS, testNS).UID(testUID).Obj(),
@@ -82,6 +84,7 @@ func TestReconciler(t *testing.T) {
 							Count:           1,
 							TopologyRequest: nil,
 						}).
+					Priority(0).
 					Obj(),
 			},
 			wantEvents: []utiltesting.EventRecord{
@@ -150,6 +153,7 @@ func TestReconciler(t *testing.T) {
 							TopologyRequest: nil,
 						},
 					).
+					Priority(0).
 					Obj(),
 			},
 			wantEvents: []utiltesting.EventRecord{
@@ -220,6 +224,7 @@ func TestReconciler(t *testing.T) {
 							TopologyRequest: nil,
 						},
 					).
+					Priority(0).
 					Obj(),
 				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "1"), testNS).
 					Annotation(podconstants.IsGroupWorkloadAnnotationKey, podconstants.IsGroupWorkloadAnnotationValue).
@@ -250,6 +255,7 @@ func TestReconciler(t *testing.T) {
 							TopologyRequest: nil,
 						},
 					).
+					Priority(0).
 					Obj(),
 			},
 			wantEvents: []utiltesting.EventRecord{
@@ -367,6 +373,54 @@ func TestReconciler(t *testing.T) {
 							},
 						},
 					).
+					Priority(0).
+					Obj(),
+			},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Name: testLWS, Namespace: testNS},
+					EventType: corev1.EventTypeNormal,
+					Reason:    jobframework.ReasonCreatedWorkload,
+					Message: fmt.Sprintf(
+						"Created Workload: %s/%s",
+						testNS,
+						GetWorkloadName(types.UID(testUID), testLWS, "0"),
+					),
+				},
+			},
+		},
+		"should create prebuilt workload with workload priority": {
+			leaderWorkerSet: leaderworkerset.MakeLeaderWorkerSet(testLWS, testNS).
+				WorkloadPriorityClass("high-priority").
+				UID(testUID).
+				Obj(),
+			workloadPriorityClasses: []kueue.WorkloadPriorityClass{
+				*utiltesting.MakeWorkloadPriorityClass("high-priority").PriorityValue(5000).Obj(),
+			},
+			wantLeaderWorkerSet: leaderworkerset.MakeLeaderWorkerSet(testLWS, testNS).
+				WorkloadPriorityClass("high-priority").
+				UID(testUID).
+				Obj(),
+			wantWorkloads: []kueue.Workload{
+				*utiltesting.MakeWorkload(GetWorkloadName(types.UID(testUID), testLWS, "0"), testNS).
+					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,
+							TopologyRequest: nil,
+						}).
+					PriorityClass("high-priority").
+					Priority(5000).
+					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Obj(),
 			},
 			wantEvents: []utiltesting.EventRecord{
@@ -388,11 +442,14 @@ func TestReconciler(t *testing.T) {
 			ctx, _ := utiltesting.ContextWithLog(t)
 			clientBuilder := utiltesting.NewClientBuilder(leaderworkersetv1.AddToScheme)
 
-			objs := make([]client.Object, 0, len(tc.workloads)+1)
+			objs := make([]client.Object, 0, len(tc.workloads)+len(tc.workloadPriorityClasses)+1)
 			objs = append(objs, tc.leaderWorkerSet)
 			for _, wl := range tc.workloads {
 				objs = append(objs, &wl)
 			}
+			for _, wpc := range tc.workloadPriorityClasses {
+				objs = append(objs, &wpc)
+			}
 
 			kClient := clientBuilder.WithObjects(objs...).Build()
 			recorder := &utiltesting.EventRecorder{}
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
index f44080766..2b957e7a6 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
@@ -49,6 +49,25 @@ func TestDefault(t *testing.T) {
 		enableIntegrations         []string
 		want                       *leaderworkersetv1.LeaderWorkerSet
 	}{
+		"LeaderWorkerSet with WorkloadPriorityClass": {
+			localQueueDefaulting: true,
+			defaultLqExist:       true,
+			lws: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "default").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				WorkloadPriorityClass("high-priority").
+				Obj(),
+			want: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "default").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("default").
+				WorkloadPriorityClass("high-priority").
+				LeaderTemplateSpecLabel(constants.WorkloadPriorityClassLabel, "high-priority").
+				LeaderTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				LeaderTemplateSpecAnnotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
+				WorkerTemplateSpecLabel(constants.WorkloadPriorityClassLabel, "high-priority").
+				WorkerTemplateSpecAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
+				WorkerTemplateSpecAnnotation(podconstants.GroupServingAnnotationKey, podconstants.GroupServingAnnotationValue).
+				Obj(),
+		},
 		"LocalQueueDefaulting enabled, default lq is created, job doesn't have queue label": {
 			localQueueDefaulting: true,
 			defaultLqExist:       true,
diff --git a/pkg/util/testingjobs/leaderworkerset/wrappers.go b/pkg/util/testingjobs/leaderworkerset/wrappers.go
index 4d9efe332..cc0b620d2 100644
--- a/pkg/util/testingjobs/leaderworkerset/wrappers.go
+++ b/pkg/util/testingjobs/leaderworkerset/wrappers.go
@@ -130,6 +130,15 @@ func (w *LeaderWorkerSetWrapper) WorkerTemplateSpecQueue(q string) *LeaderWorker
 	return w.WorkerTemplateSpecLabel(constants.QueueLabel, q)
 }
 
+// LeaderTemplateSpecLabel sets the label of the pod template spec of the LeaderLeaderSet
+func (w *LeaderWorkerSetWrapper) LeaderTemplateSpecLabel(k, v string) *LeaderWorkerSetWrapper {
+	if w.Spec.LeaderWorkerTemplate.LeaderTemplate.Labels == nil {
+		w.Spec.LeaderWorkerTemplate.LeaderTemplate.Labels = make(map[string]string, 1)
+	}
+	w.Spec.LeaderWorkerTemplate.LeaderTemplate.Labels[k] = v
+	return w
+}
+
 // LeaderTemplateSpecAnnotation sets the annotation of the pod template spec of the LeaderLeaderSet
 func (w *LeaderWorkerSetWrapper) LeaderTemplateSpecAnnotation(k, v string) *LeaderWorkerSetWrapper {
 	if w.Spec.LeaderWorkerTemplate.LeaderTemplate.Annotations == nil {
@@ -210,3 +219,8 @@ func (w *LeaderWorkerSetWrapper) TerminationGracePeriod(seconds int64) *LeaderWo
 	w.Spec.LeaderWorkerTemplate.WorkerTemplate.Spec.TerminationGracePeriodSeconds = &seconds
 	return w
 }
+
+// WorkloadPriorityClass sets workloadpriorityclass.
+func (w *LeaderWorkerSetWrapper) WorkloadPriorityClass(wpc string) *LeaderWorkerSetWrapper {
+	return w.Label(constants.WorkloadPriorityClassLabel, wpc)
+}
diff --git a/test/e2e/singlecluster/leaderworkerset_test.go b/test/e2e/singlecluster/leaderworkerset_test.go
index 2c28ac60b..373b1e4fa 100644
--- a/test/e2e/singlecluster/leaderworkerset_test.go
+++ b/test/e2e/singlecluster/leaderworkerset_test.go
@@ -28,6 +28,7 @@ import (
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/leaderworkerset"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	leaderworkersettesting "sigs.k8s.io/kueue/pkg/util/testingjobs/leaderworkerset"
@@ -660,4 +661,134 @@ var _ = ginkgo.Describe("LeaderWorkerSet integration", func() {
 			})
 		})
 	})
+
+	ginkgo.When("LeaderWorkerSet created with WorkloadPriorityClass", func() {
+		var (
+			highPriorityWPC *kueue.WorkloadPriorityClass
+			lowPriorityWPC  *kueue.WorkloadPriorityClass
+		)
+
+		ginkgo.BeforeEach(func() {
+			highPriorityWPC = testing.MakeWorkloadPriorityClass("high-priority").
+				PriorityValue(5000).
+				Obj()
+			gomega.Expect(k8sClient.Create(ctx, highPriorityWPC)).To(gomega.Succeed())
+
+			lowPriorityWPC = testing.MakeWorkloadPriorityClass("low-priority").
+				PriorityValue(1000).
+				Obj()
+			gomega.Expect(k8sClient.Create(ctx, lowPriorityWPC)).To(gomega.Succeed())
+		})
+
+		ginkgo.AfterEach(func() {
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, highPriorityWPC, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, lowPriorityWPC, true)
+		})
+
+		ginkgo.It("should allow to update the PodTemplate in LeaderWorkerSet", func() {
+			lowPriorityLWS := leaderworkersettesting.MakeLeaderWorkerSet("low-priority", ns.Name).
+				Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+				Size(3).
+				Replicas(1).
+				RequestAndLimit(corev1.ResourceCPU, "1").
+				TerminationGracePeriod(1).
+				Queue(lq.Name).
+				WorkloadPriorityClass(lowPriorityWPC.Name).
+				Obj()
+			ginkgo.By("Create a low priority LeaderWorkerSet", func() {
+				gomega.Expect(k8sClient.Create(ctx, lowPriorityLWS)).To(gomega.Succeed())
+			})
+
+			ginkgo.By("Waiting for replicas is ready in low priority LeaderWorkerSet", func() {
+				createdLowPriorityLWS := &leaderworkersetv1.LeaderWorkerSet{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lowPriorityLWS), createdLowPriorityLWS)).To(gomega.Succeed())
+					g.Expect(createdLowPriorityLWS.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdLowPriorityLWS.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())
+			})
+
+			createdLowPriorityWl := &kueue.Workload{}
+			lowPriorityWlLookupKey := types.NamespacedName{
+				Name:      leaderworkerset.GetWorkloadName(lowPriorityLWS.UID, lowPriorityLWS.Name, "0"),
+				Namespace: ns.Name,
+			}
+			ginkgo.By("Check the low priority Workload is created and admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, lowPriorityWlLookupKey, createdLowPriorityWl)).To(gomega.Succeed())
+					g.Expect(createdLowPriorityWl.Spec.PriorityClassSource).To(gomega.Equal(constants.WorkloadPriorityClassSource))
+					g.Expect(createdLowPriorityWl.Spec.PriorityClassName).To(gomega.Equal(lowPriorityWPC.Name))
+					g.Expect(createdLowPriorityWl.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			highPriorityLWS := leaderworkersettesting.MakeLeaderWorkerSet("high-priority", ns.Name).
+				Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+				Size(3).
+				Replicas(1).
+				RequestAndLimit(corev1.ResourceCPU, "1").
+				TerminationGracePeriod(1).
+				Queue(lq.Name).
+				WorkloadPriorityClass(highPriorityWPC.Name).
+				Obj()
+			ginkgo.By("Create a high priority LeaderWorkerSet", func() {
+				gomega.Expect(k8sClient.Create(ctx, highPriorityLWS)).To(gomega.Succeed())
+			})
+
+			ginkgo.By("Waiting for ready replicas in the high priority LeaderWorkerSet", func() {
+				createdHighPriorityLWS := &leaderworkersetv1.LeaderWorkerSet{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(highPriorityLWS), createdHighPriorityLWS)).To(gomega.Succeed())
+					g.Expect(createdHighPriorityLWS.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdHighPriorityLWS.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("Await for the low-priory LeaderWorkerSet to be preempted", func() {
+				createdLowPriorityLWS := &leaderworkersetv1.LeaderWorkerSet{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lowPriorityLWS), createdLowPriorityLWS)).To(gomega.Succeed())
+					g.Expect(createdLowPriorityLWS.Status.ReadyReplicas).To(gomega.Equal(int32(0)))
+					g.Expect(createdLowPriorityLWS.Status.Conditions).To(gomega.ContainElements(
+						gomega.BeComparableTo(metav1.Condition{
+							Type:   "Progressing",
+							Status: metav1.ConditionTrue,
+							Reason: "GroupsProgressing",
+						}, util.IgnoreConditionTimestampsAndObservedGeneration, util.IgnoreConditionMessage),
+						gomega.BeComparableTo(metav1.Condition{
+							Type:   "Available",
+							Status: metav1.ConditionFalse,
+							Reason: "AllGroupsReady",
+						}, util.IgnoreConditionTimestampsAndObservedGeneration, util.IgnoreConditionMessage),
+					))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			createdHighPriorityWl := &kueue.Workload{}
+			highPriorityWlLookupKey := types.NamespacedName{
+				Name:      leaderworkerset.GetWorkloadName(highPriorityLWS.UID, highPriorityLWS.Name, "0"),
+				Namespace: ns.Name,
+			}
+			ginkgo.By("Await for the high priority Workload to be admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, highPriorityWlLookupKey, createdHighPriorityWl)).To(gomega.Succeed())
+					g.Expect(createdHighPriorityWl.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+					g.Expect(createdHighPriorityWl.Spec.PriorityClassSource).To(gomega.Equal(constants.WorkloadPriorityClassSource))
+					g.Expect(createdHighPriorityWl.Spec.PriorityClassName).To(gomega.Equal(highPriorityLWS.Name))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
 })