Repository: kubernetes-sigs/kueue

Issue:
Support mutating workload priority class while running without disruption
**What would you like to be added**:

Support mutating the workload priority class label. 

Currently it is blocked here: https://github.com/kubernetes-sigs/kueue/blob/15ea4e843157cb86b443ad8ff0f5ebeed9327ef2/pkg/controller/jobframework/validation.go#L151-L157

However, it seems like a precautionary measure rather than a fundamental problem.

Once the priority is lowered we should make sure the scheduling is triggered as we may be now having a new workload that can schedule preempting the running one.

**Why is this needed**:

Sometimes users may want change priority of a running workload. They are ok if the workload gets interrupted (preempted) by the inflow of incoming workloads, but would like to avoid preemption if not needed.

Hints from issue comments:
@mimowo: cc @mwielgus @mwysokin @mbobrovskyi @mbobrovskyi: /assign@tenzen-y: I like this enhancement, which allows us to construct the budget-based workload management.
Let's say that we have a billing and usage management system separate from Kueue.
The external costom controllers watches the budget (billing) monitoring system, then it automatically change the WorkloadPriorityClass if the tenants use the quota with exceeding the pre-defined ones in a external Custom Resource.


Solution patch:
diff --git a/apis/kueue/v1beta1/workload_types.go b/apis/kueue/v1beta1/workload_types.go
index 7d297f96f..0a6cb4e10 100644
--- a/apis/kueue/v1beta1/workload_types.go
+++ b/apis/kueue/v1beta1/workload_types.go
@@ -766,7 +766,7 @@ const (
 // Workload is the Schema for the workloads API
 // +kubebuilder:validation:XValidation:rule="has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True') && has(self.status.admission) ? size(self.spec.podSets) == size(self.status.admission.podSetAssignments) : true", message="podSetAssignments must have the same number of podSets as the spec"
 // +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True')) ? (oldSelf.spec.priorityClassSource == self.spec.priorityClassSource) : true", message="priorityClassSource is immutable while workload quota reserved"
-// +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true", message="priorityClassName is immutable while workload quota reserved"
+// +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True') && (self.spec.priorityClassSource != 'kueue.x-k8s.io/workloadpriorityclass') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true", message="priorityClassName is immutable while workload quota reserved and priorityClassSource is not equal to kueue.x-k8s.io/workloadpriorityclass"
 // +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True')) && (has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True')) && has(oldSelf.spec.queueName) && has(self.spec.queueName) ? oldSelf.spec.queueName == self.spec.queueName : true", message="queueName is immutable while workload quota reserved"
 // +kubebuilder:validation:XValidation:rule="((has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'Admitted' && c.status == 'True')) && (has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == 'Admitted' && c.status == 'True')))?((has(oldSelf.spec.maximumExecutionTimeSeconds)?oldSelf.spec.maximumExecutionTimeSeconds:0) ==  (has(self.spec.maximumExecutionTimeSeconds)?self.spec.maximumExecutionTimeSeconds:0)):true", message="maximumExecutionTimeSeconds is immutable while workload quota reserved"
 type Workload struct {
diff --git a/apis/kueue/v1beta2/workload_types.go b/apis/kueue/v1beta2/workload_types.go
index 5010fa0e0..623ed4fe8 100644
--- a/apis/kueue/v1beta2/workload_types.go
+++ b/apis/kueue/v1beta2/workload_types.go
@@ -765,7 +765,7 @@ const (
 // Workload is the Schema for the workloads API
 // +kubebuilder:validation:XValidation:rule="has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True') && has(self.status.admission) ? size(self.spec.podSets) == size(self.status.admission.podSetAssignments) : true", message="podSetAssignments must have the same number of podSets as the spec"
 // +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True')) ? ((!has(self.spec.priorityClassSource) && !has(oldSelf.spec.priorityClassSource)) || (has(self.spec.priorityClassSource) && has(oldSelf.spec.priorityClassSource) && self.spec.priorityClassSource == oldSelf.spec.priorityClassSource) || (!has(self.spec.priorityClassSource) && has(oldSelf.spec.priorityClassSource) && oldSelf.spec.priorityClassSource.size() == 0) || (!has(oldSelf.spec.priorityClassSource) && has(self.spec.priorityClassSource) && self.spec.priorityClassSource.size() == 0)) : true", message="priorityClassSource is immutable while workload quota reserved"
-// +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true",message="priorityClassName is immutable while workload quota reserved"
+// +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True') && (self.spec.priorityClassSource != 'kueue.x-k8s.io/workloadpriorityclass') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true", message="priorityClassName is immutable while workload quota reserved and priorityClassSource is not equal to kueue.x-k8s.io/workloadpriorityclass"
 // +kubebuilder:validation:XValidation:rule="(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True')) && (has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status == 'True')) && has(oldSelf.spec.queueName) && has(self.spec.queueName) ? oldSelf.spec.queueName == self.spec.queueName : true", message="queueName is immutable while workload quota reserved"
 // +kubebuilder:validation:XValidation:rule="((has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == 'Admitted' && c.status == 'True')) && (has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == 'Admitted' && c.status == 'True')))?((has(oldSelf.spec.maximumExecutionTimeSeconds)?oldSelf.spec.maximumExecutionTimeSeconds:0) ==  (has(self.spec.maximumExecutionTimeSeconds)?self.spec.maximumExecutionTimeSeconds:0)):true", message="maximumExecutionTimeSeconds is immutable while workload quota reserved"
 type Workload struct {
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
index da71f3d5f..fd32cadfa 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
@@ -8846,8 +8846,8 @@ spec:
               rule: 'has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'') && has(self.status.admission) ? size(self.spec.podSets) == size(self.status.admission.podSetAssignments) : true'
             - message: priorityClassSource is immutable while workload quota reserved
               rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'')) ? (oldSelf.spec.priorityClassSource == self.spec.priorityClassSource) : true'
-            - message: priorityClassName is immutable while workload quota reserved
-              rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true'
+            - message: priorityClassName is immutable while workload quota reserved and priorityClassSource is not equal to kueue.x-k8s.io/workloadpriorityclass
+              rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'') && (self.spec.priorityClassSource != ''kueue.x-k8s.io/workloadpriorityclass'') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true'
             - message: queueName is immutable while workload quota reserved
               rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'')) && (has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'')) && has(oldSelf.spec.queueName) && has(self.spec.queueName) ? oldSelf.spec.queueName == self.spec.queueName : true'
             - message: maximumExecutionTimeSeconds is immutable while workload quota reserved
@@ -17670,8 +17670,8 @@ spec:
               rule: 'has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'') && has(self.status.admission) ? size(self.spec.podSets) == size(self.status.admission.podSetAssignments) : true'
             - message: priorityClassSource is immutable while workload quota reserved
               rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'')) ? ((!has(self.spec.priorityClassSource) && !has(oldSelf.spec.priorityClassSource)) || (has(self.spec.priorityClassSource) && has(oldSelf.spec.priorityClassSource) && self.spec.priorityClassSource == oldSelf.spec.priorityClassSource) || (!has(self.spec.priorityClassSource) && has(oldSelf.spec.priorityClassSource) && oldSelf.spec.priorityClassSource.size() == 0) || (!has(oldSelf.spec.priorityClassSource) && has(self.spec.priorityClassSource) && self.spec.priorityClassSource.size() == 0)) : true'
-            - message: priorityClassName is immutable while workload quota reserved
-              rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true'
+            - message: priorityClassName is immutable while workload quota reserved and priorityClassSource is not equal to kueue.x-k8s.io/workloadpriorityclass
+              rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'') && (self.spec.priorityClassSource != ''kueue.x-k8s.io/workloadpriorityclass'') && has(oldSelf.spec.priorityClassName) && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName == self.spec.priorityClassName) : true'
             - message: queueName is immutable while workload quota reserved
               rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'')) && (has(self.status) && has(self.status.conditions) && self.status.conditions.exists(c, c.type == ''QuotaReserved'' && c.status == ''True'')) && has(oldSelf.spec.queueName) && has(self.spec.queueName) ? oldSelf.spec.queueName == self.spec.queueName : true'
             - message: maximumExecutionTimeSeconds is immutable while workload quota reserved
diff --git a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
index e83726674..4ed717e21 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
@@ -9345,9 +9345,11 @@ spec:
           rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c,
             c.type == ''QuotaReserved'' && c.status == ''True'')) ? (oldSelf.spec.priorityClassSource
             == self.spec.priorityClassSource) : true'
-        - message: priorityClassName is immutable while workload quota reserved
+        - message: priorityClassName is immutable while workload quota reserved and
+            priorityClassSource is not equal to kueue.x-k8s.io/workloadpriorityclass
           rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c,
-            c.type == ''QuotaReserved'' && c.status == ''True'') && has(oldSelf.spec.priorityClassName)
+            c.type == ''QuotaReserved'' && c.status == ''True'') && (self.spec.priorityClassSource
+            != ''kueue.x-k8s.io/workloadpriorityclass'') && has(oldSelf.spec.priorityClassName)
             && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName
             == self.spec.priorityClassName) : true'
         - message: queueName is immutable while workload quota reserved
@@ -18701,9 +18703,11 @@ spec:
             && has(oldSelf.spec.priorityClassSource) && oldSelf.spec.priorityClassSource.size()
             == 0) || (!has(oldSelf.spec.priorityClassSource) && has(self.spec.priorityClassSource)
             && self.spec.priorityClassSource.size() == 0)) : true'
-        - message: priorityClassName is immutable while workload quota reserved
+        - message: priorityClassName is immutable while workload quota reserved and
+            priorityClassSource is not equal to kueue.x-k8s.io/workloadpriorityclass
           rule: '(has(oldSelf.status) && has(oldSelf.status.conditions) && oldSelf.status.conditions.exists(c,
-            c.type == ''QuotaReserved'' && c.status == ''True'') && has(oldSelf.spec.priorityClassName)
+            c.type == ''QuotaReserved'' && c.status == ''True'') && (self.spec.priorityClassSource
+            != ''kueue.x-k8s.io/workloadpriorityclass'') && has(oldSelf.spec.priorityClassName)
             && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName
             == self.spec.priorityClassName) : true'
         - message: queueName is immutable while workload quota reserved
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index eeee91a2c..b248d6ce5 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -52,6 +52,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
 	qcache "sigs.k8s.io/kueue/pkg/cache/queue"
 	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
+	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
 	"sigs.k8s.io/kueue/pkg/dra"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -898,7 +899,8 @@ func (r *WorkloadReconciler) Update(e event.TypedUpdateEvent[*kueue.Workload]) b
 			}
 		}
 	case prevStatus == workload.StatusAdmitted && status == workload.StatusAdmitted && !equality.Semantic.DeepEqual(e.ObjectOld.Status.ReclaimablePods, e.ObjectNew.Status.ReclaimablePods),
-		features.Enabled(features.ElasticJobsViaWorkloadSlices) && workloadslicing.ScaledDown(workload.ExtractPodSetCountsFromWorkload(e.ObjectOld), workload.ExtractPodSetCountsFromWorkload(e.ObjectNew)):
+		features.Enabled(features.ElasticJobsViaWorkloadSlices) && workloadslicing.ScaledDown(workload.ExtractPodSetCountsFromWorkload(e.ObjectOld), workload.ExtractPodSetCountsFromWorkload(e.ObjectNew)),
+		workloadPriorityClassChanged(e.ObjectOld, e.ObjectNew):
 		// trigger the move of associated inadmissibleWorkloads, if there are any.
 		r.queues.QueueAssociatedInadmissibleWorkloadsAfter(ctx, e.ObjectNew, func() {
 			// Update the workload from cache while holding the queues lock
@@ -920,6 +922,13 @@ func (r *WorkloadReconciler) Update(e event.TypedUpdateEvent[*kueue.Workload]) b
 	return true
 }
 
+func workloadPriorityClassChanged(old, new *kueue.Workload) bool {
+	return old.Spec.PriorityClassSource == constants.WorkloadPriorityClassSource &&
+		new.Spec.PriorityClassSource == constants.WorkloadPriorityClassSource &&
+		old.Spec.PriorityClassName != "" && new.Spec.PriorityClassName != "" &&
+		old.Spec.PriorityClassName != new.Spec.PriorityClassName
+}
+
 func (r *WorkloadReconciler) Generic(e event.TypedGenericEvent[*kueue.Workload]) bool {
 	r.log.V(3).Info("Ignore Workload generic event", "workload", klog.KObj(e.Object))
 	return false
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index 19b9935a9..e1bbd1cb4 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -598,15 +598,6 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 			}
 			return ctrl.Result{}, err
 		}
-		// Update workload priority if job's label changed.
-		if wl.Spec.PriorityClassSource == constants.WorkloadPriorityClassSource && WorkloadPriorityClassName(object) != wl.Spec.PriorityClassName {
-			log.V(2).Info("Job changed priority, updating workload", "oldPriority", wl.Spec.PriorityClassName, "newPriority", WorkloadPriorityClassName(object))
-			if _, err = r.updateWorkloadToMatchJob(ctx, job, object, wl); err != nil {
-				log.Error(err, "Updating workload priority")
-				return ctrl.Result{}, err
-			}
-			return ctrl.Result{}, nil
-		}
 		log.V(3).Info("Job is suspended and workload not yet admitted by a clusterQueue, nothing to do")
 		return ctrl.Result{}, nil
 	}
@@ -934,9 +925,32 @@ func (r *JobReconciler) ensureOneWorkload(ctx context.Context, job GenericJob, o
 		return r.updateWorkloadToMatchJob(ctx, job, object, toUpdate)
 	}
 
+	if match != nil {
+		if err := UpdateWorkloadPriority(ctx, r.client, r.record, job.Object(), match, getCustomPriorityClassFuncFromJob(job)); err != nil {
+			return nil, err
+		}
+	}
+
 	return match, nil
 }
 
+// UpdateWorkloadPriority updates workload priority if object's kueue.x-k8s.io/priority-class label changed.
+func UpdateWorkloadPriority(ctx context.Context, c client.Client, r record.EventRecorder, obj client.Object, wl *kueue.Workload, customPriorityClassFunc func() string) error {
+	if wl.Spec.PriorityClassSource == constants.WorkloadPriorityClassSource && WorkloadPriorityClassName(obj) != wl.Spec.PriorityClassName {
+		if err := PrepareWorkloadPriority(ctx, c, obj, wl, customPriorityClassFunc); err != nil {
+			return fmt.Errorf("prepare workload priority: %w", err)
+		}
+		if err := c.Update(ctx, wl); err != nil {
+			return fmt.Errorf("updating existing workload: %w", err)
+		}
+		r.Eventf(obj,
+			corev1.EventTypeNormal, ReasonUpdatedWorkload,
+			"Updated workload priority class: %v", klog.KObj(wl),
+		)
+	}
+	return nil
+}
+
 func FindMatchingWorkloads(ctx context.Context, c client.Client, job GenericJob) (match *kueue.Workload, toDelete []*kueue.Workload, err error) {
 	object := job.Object()
 
@@ -1090,7 +1104,7 @@ func (r *JobReconciler) updateWorkloadToMatchJob(ctx context.Context, job Generi
 
 	r.record.Eventf(object, corev1.EventTypeNormal, ReasonUpdatedWorkload,
 		"Updated not matching Workload for suspended job: %v", klog.KObj(wl))
-	return newWl, nil
+	return wl, nil
 }
 
 // startJob will unsuspend the job, and also inject the node affinity.
@@ -1263,9 +1277,15 @@ func prepareWorkloadSlice(ctx context.Context, clnt client.Client, job GenericJo
 	}
 }
 
-// prepareWorkload adds the priority information for the constructed workload
-func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl *kueue.Workload) error {
-	priorityClassName, source, p, err := r.extractPriority(ctx, wl.Spec.PodSets, job)
+func getCustomPriorityClassFuncFromJob(job GenericJob) func() string {
+	if jobWithPriorityClass, isImplemented := job.(JobWithPriorityClass); isImplemented {
+		return jobWithPriorityClass.PriorityClass
+	}
+	return nil
+}
+
+func PrepareWorkloadPriority(ctx context.Context, c client.Client, obj client.Object, wl *kueue.Workload, customPriorityClassFunc func() string) error {
+	priorityClassName, source, p, err := ExtractPriority(ctx, c, obj, wl.Spec.PodSets, customPriorityClassFunc)
 	if err != nil {
 		return err
 	}
@@ -1273,6 +1293,16 @@ func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl
 	wl.Spec.PriorityClassName = priorityClassName
 	wl.Spec.Priority = &p
 	wl.Spec.PriorityClassSource = source
+
+	return nil
+}
+
+// prepareWorkload adds the priority information for the constructed workload
+func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl *kueue.Workload) error {
+	if err := PrepareWorkloadPriority(ctx, r.client, job.Object(), wl, getCustomPriorityClassFuncFromJob(job)); err != nil {
+		return err
+	}
+
 	wl.Spec.PodSets = clearMinCountsIfFeatureDisabled(wl.Spec.PodSets)
 
 	if workloadSliceEnabled(job) {
@@ -1281,20 +1311,12 @@ func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl
 	return nil
 }
 
-func (r *JobReconciler) extractPriority(ctx context.Context, podSets []kueue.PodSet, job GenericJob) (string, string, int32, error) {
-	var customPriorityFunc func() string
-	if jobWithPriorityClass, isImplemented := job.(JobWithPriorityClass); isImplemented {
-		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) {
+func ExtractPriority(ctx context.Context, c client.Client, obj client.Object, podSets []kueue.PodSet, customPriorityClassFunc 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())
+	if customPriorityClassFunc != nil {
+		return utilpriority.GetPriorityFromPriorityClass(ctx, c, customPriorityClassFunc())
 	}
 	return utilpriority.GetPriorityFromPriorityClass(ctx, c, extractPriorityFromPodSets(podSets))
 }
diff --git a/pkg/controller/jobframework/validation.go b/pkg/controller/jobframework/validation.go
index a3a94cb7c..07b30bbbb 100644
--- a/pkg/controller/jobframework/validation.go
+++ b/pkg/controller/jobframework/validation.go
@@ -148,11 +148,7 @@ func validateUpdateForPrebuiltWorkload(oldJob, newJob GenericJob) field.ErrorLis
 }
 
 func validateJobUpdateForWorkloadPriorityClassName(oldJob, newJob GenericJob) field.ErrorList {
-	var allErrs field.ErrorList
-	if !newJob.IsSuspended() || IsWorkloadPriorityClassNameEmpty(newJob.Object()) {
-		allErrs = append(allErrs, ValidateUpdateForWorkloadPriorityClassName(oldJob.Object(), newJob.Object())...)
-	}
-	return allErrs
+	return ValidateUpdateForWorkloadPriorityClassName(newJob.IsSuspended(), oldJob.Object(), newJob.Object())
 }
 
 // validatedUpdateForEnabledWorkloadSlice validates that the workload-slicing toggle remains immutable on update.
@@ -168,9 +164,11 @@ func validatedUpdateForEnabledWorkloadSlice(oldJob, newJob GenericJob) field.Err
 	return nil
 }
 
-func ValidateUpdateForWorkloadPriorityClassName(oldObj, newObj client.Object) field.ErrorList {
-	allErrs := apivalidation.ValidateImmutableField(WorkloadPriorityClassName(newObj), WorkloadPriorityClassName(oldObj), workloadPriorityClassNamePath)
-	return allErrs
+func ValidateUpdateForWorkloadPriorityClassName(isSuspended bool, oldObj, newObj client.Object) field.ErrorList {
+	if !isSuspended && IsWorkloadPriorityClassNameEmpty(oldObj) || IsWorkloadPriorityClassNameEmpty(newObj) {
+		return apivalidation.ValidateImmutableField(WorkloadPriorityClassName(newObj), WorkloadPriorityClassName(oldObj), workloadPriorityClassNamePath)
+	}
+	return nil
 }
 
 func validateCreateForMaxExecTime(job GenericJob) field.ErrorList {
diff --git a/pkg/controller/jobs/deployment/deployment_webhook.go b/pkg/controller/jobs/deployment/deployment_webhook.go
index 5a2b71b59..f47abad41 100644
--- a/pkg/controller/jobs/deployment/deployment_webhook.go
+++ b/pkg/controller/jobs/deployment/deployment_webhook.go
@@ -133,9 +133,11 @@ func (wh *Webhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Ob
 	if !isSuspended || newQueueName == "" {
 		allErrs = append(allErrs, apivalidation.ValidateImmutableField(newQueueName, oldQueueName, queueNameLabelPath)...)
 	}
-	if !isSuspended || jobframework.IsWorkloadPriorityClassNameEmpty(newDeployment.Object()) {
-		allErrs = append(allErrs, jobframework.ValidateUpdateForWorkloadPriorityClassName(oldDeployment.Object(), newDeployment.Object())...)
-	}
+	allErrs = append(allErrs, jobframework.ValidateUpdateForWorkloadPriorityClassName(
+		isSuspended,
+		oldDeployment.Object(),
+		newDeployment.Object(),
+	)...)
 	return warnings, allErrs.ToAggregate()
 }
 
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
index 70ad6b64a..615dff331 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
@@ -106,7 +106,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 		return ctrl.Result{}, err
 	}
 
-	toCreate, toFinalize := r.filterWorkloads(lws, wlList.Items)
+	toCreate, toUpdate, toFinalize := r.filterWorkloads(lws, wlList.Items)
 
 	eg, ctx := errgroup.WithContext(ctx)
 
@@ -116,6 +116,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 		})
 	})
 
+	eg.Go(func() error {
+		return parallelize.Until(ctx, len(toUpdate), func(i int) error {
+			return jobframework.UpdateWorkloadPriority(ctx, r.client, r.record, lws, toUpdate[i], nil)
+		})
+	})
+
 	eg.Go(func() error {
 		return parallelize.Until(ctx, len(toFinalize), func(i int) error {
 			return r.removeOwnerReference(ctx, lws, toFinalize[i])
@@ -130,15 +136,17 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	return ctrl.Result{}, nil
 }
 
-// filterWorkloads compares the desired state in a LeaderWorkerSet with existing workloads,
-// identifying workloads to create and those to finalize.
+// filterWorkloads compares the desired state of a LeaderWorkerSet with existing workloads,
+// determining which workloads need to be created, updated, or finalized.
 //
-// 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) {
+// It accepts a LeaderWorkerSet and a slice of existing Workload objects as input and returns:
+// 1. A slice of workload names to be created
+// 2. A slice of workloads that may require updates
+// 3. A slice of Workload pointers to be finalized
+func (r *Reconciler) filterWorkloads(lws *leaderworkersetv1.LeaderWorkerSet, existingWorkloads []kueue.Workload) ([]string, []*kueue.Workload, []*kueue.Workload) {
 	var (
 		toCreate   []string
+		toUpdate   []*kueue.Workload
 		toFinalize = utilslices.ToRefMap(existingWorkloads, func(e *kueue.Workload) string {
 			return e.Name
 		})
@@ -147,14 +155,15 @@ func (r *Reconciler) filterWorkloads(lws *leaderworkersetv1.LeaderWorkerSet, exi
 
 	for i := range replicas {
 		workloadName := GetWorkloadName(lws.UID, lws.Name, fmt.Sprint(i))
-		if _, ok := toFinalize[workloadName]; ok {
+		if wl, ok := toFinalize[workloadName]; ok {
+			toUpdate = append(toUpdate, wl)
 			delete(toFinalize, workloadName)
 		} else {
 			toCreate = append(toCreate, workloadName)
 		}
 	}
 
-	return toCreate, slices.Collect(maps.Values(toFinalize))
+	return toCreate, toUpdate, slices.Collect(maps.Values(toFinalize))
 }
 
 func (r *Reconciler) createPrebuiltWorkload(ctx context.Context, lws *leaderworkersetv1.LeaderWorkerSet, workloadName string) error {
@@ -163,15 +172,11 @@ func (r *Reconciler) createPrebuiltWorkload(ctx context.Context, lws *leaderwork
 		return err
 	}
 
-	priorityClassName, source, p, err := jobframework.ExtractPriority(ctx, r.client, lws, createdWorkload.Spec.PodSets, nil)
+	err = jobframework.PrepareWorkloadPriority(ctx, r.client, lws, createdWorkload, 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 4a2a51c77..917404b0a 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
@@ -156,12 +156,11 @@ func (wh *Webhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Ob
 	)...)
 
 	isSuspended := oldLeaderWorkerSet.Status.ReadyReplicas == 0
-	if !isSuspended || jobframework.IsWorkloadPriorityClassNameEmpty(newLeaderWorkerSet.Object()) {
-		allErrs = append(allErrs, jobframework.ValidateUpdateForWorkloadPriorityClassName(
-			newLeaderWorkerSet.Object(),
-			oldLeaderWorkerSet.Object(),
-		)...)
-	}
+	allErrs = append(allErrs, jobframework.ValidateUpdateForWorkloadPriorityClassName(
+		isSuspended,
+		oldLeaderWorkerSet.Object(),
+		newLeaderWorkerSet.Object(),
+	)...)
 
 	suspend, err := jobframework.WorkloadShouldBeSuspended(ctx, newLeaderWorkerSet.Object(), wh.client, wh.manageJobsWithoutQueueName, wh.managedJobsNamespaceSelector)
 	if err != nil {
diff --git a/pkg/controller/jobs/statefulset/statefulset_webhook.go b/pkg/controller/jobs/statefulset/statefulset_webhook.go
index 6a7aa4ee0..0b9e36c8f 100644
--- a/pkg/controller/jobs/statefulset/statefulset_webhook.go
+++ b/pkg/controller/jobs/statefulset/statefulset_webhook.go
@@ -144,12 +144,11 @@ func (wh *Webhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Ob
 	if !isSuspended || newQueueName == "" {
 		allErrs = append(allErrs, apivalidation.ValidateImmutableField(newQueueName, oldQueueName, queueNameLabelPath)...)
 	}
-	if !isSuspended || jobframework.IsWorkloadPriorityClassNameEmpty(newStatefulSet.Object()) {
-		allErrs = append(allErrs, jobframework.ValidateUpdateForWorkloadPriorityClassName(
-			oldStatefulSet.Object(),
-			newStatefulSet.Object(),
-		)...)
-	}
+	allErrs = append(allErrs, jobframework.ValidateUpdateForWorkloadPriorityClassName(
+		isSuspended,
+		oldStatefulSet.Object(),
+		newStatefulSet.Object(),
+	)...)
 
 	suspend, err := jobframework.WorkloadShouldBeSuspended(ctx, newStatefulSet.Object(), wh.client, wh.manageJobsWithoutQueueName, wh.managedJobsNamespaceSelector)
 	if err != nil {



Test patch:
diff --git a/pkg/controller/jobs/deployment/deployment_webhook_test.go b/pkg/controller/jobs/deployment/deployment_webhook_test.go
index 75cc8dc33..7b658f629 100644
--- a/pkg/controller/jobs/deployment/deployment_webhook_test.go
+++ b/pkg/controller/jobs/deployment/deployment_webhook_test.go
@@ -302,6 +302,23 @@ func TestValidateUpdate(t *testing.T) {
 				Label(constants.WorkloadPriorityClassLabel, "new-test").
 				Obj(),
 		},
+		"set priority-class when replicas ready": {
+			oldDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			newDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			wantErr: field.ErrorList{
+				&field.Error{
+					Type:  field.ErrorTypeInvalid,
+					Field: "metadata.labels[kueue.x-k8s.io/priority-class]",
+				},
+			}.ToAggregate(),
+		},
 		"update priority-class when replicas ready": {
 			oldDeployment: testingdeployment.MakeDeployment("test-pod", "").
 				Queue("test-queue").
@@ -313,6 +330,54 @@ func TestValidateUpdate(t *testing.T) {
 				Label(constants.WorkloadPriorityClassLabel, "new-test").
 				ReadyReplicas(int32(1)).
 				Obj(),
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"delete priority-class when replicas ready": {
+			oldDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				ReadyReplicas(int32(1)).
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				Obj(),
+			newDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			wantErr: field.ErrorList{
+				&field.Error{
+					Type:  field.ErrorTypeInvalid,
+					Field: "metadata.labels[kueue.x-k8s.io/priority-class]",
+				},
+			}.ToAggregate(),
+		},
+		"set priority-class when replicas not ready": {
+			oldDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				Obj(),
+			newDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				Obj(),
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"update priority-class when replicas not ready": {
+			oldDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				Obj(),
+			newDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "new-test").
+				Obj(),
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"delete priority-class when replicas not ready": {
+			oldDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				Obj(),
+			newDeployment: testingdeployment.MakeDeployment("test-pod", "").
+				Queue("test-queue").
+				Obj(),
 			wantErr: field.ErrorList{
 				&field.Error{
 					Type:  field.ErrorTypeInvalid,
diff --git a/pkg/controller/jobs/job/job_controller_test.go b/pkg/controller/jobs/job/job_controller_test.go
index b5879f272..c734bcf34 100644
--- a/pkg/controller/jobs/job/job_controller_test.go
+++ b/pkg/controller/jobs/job/job_controller_test.go
@@ -2714,7 +2714,7 @@ func TestReconciler(t *testing.T) {
 					Key:       types.NamespacedName{Name: "job", Namespace: "ns"},
 					EventType: "Normal",
 					Reason:    "UpdatedWorkload",
-					Message:   "Updated not matching Workload for suspended job: ns/job",
+					Message:   "Updated workload priority class: ns/job",
 				},
 			},
 		},
diff --git a/pkg/controller/jobs/job/job_webhook_test.go b/pkg/controller/jobs/job/job_webhook_test.go
index 5c4e7c32c..3be566a02 100644
--- a/pkg/controller/jobs/job/job_webhook_test.go
+++ b/pkg/controller/jobs/job/job_webhook_test.go
@@ -518,17 +518,37 @@ func TestValidateUpdate(t *testing.T) {
 			wantValidationErrs: nil,
 		},
 		{
-			name:   "workloadPriorityClassName is mutable when job is suspended",
-			oldJob: testingutil.MakeJob("job", "default").WorkloadPriorityClass("test-1").Obj(),
-			newJob: testingutil.MakeJob("job", "default").WorkloadPriorityClass("test-2").Obj(),
+			name:               "set priority-class when job not suspend",
+			oldJob:             testingutil.MakeJob("job", "default").Suspend(false).Obj(),
+			newJob:             testingutil.MakeJob("job", "default").Suspend(false).WorkloadPriorityClass("test").Obj(),
+			wantValidationErrs: apivalidation.ValidateImmutableField("test", "", workloadPriorityClassNamePath),
 		},
 		{
-			name:   "workloadPriorityClassName is immutable when job is running",
-			oldJob: testingutil.MakeJob("job", "default").WorkloadPriorityClass("test-1").Suspend(false).Obj(),
-			newJob: testingutil.MakeJob("job", "default").WorkloadPriorityClass("test-2").Suspend(false).Obj(),
-			wantValidationErrs: field.ErrorList{
-				field.Invalid(workloadPriorityClassNamePath, "test-2", apivalidation.FieldImmutableErrorMsg),
-			},
+			name:   "update priority-class when job not suspend",
+			oldJob: testingutil.MakeJob("job", "default").Suspend(false).WorkloadPriorityClass("test").Obj(),
+			newJob: testingutil.MakeJob("job", "default").Suspend(false).WorkloadPriorityClass("new-test").Obj(),
+		},
+		{
+			name:               "delete priority-class when job not suspend",
+			oldJob:             testingutil.MakeJob("job", "default").Suspend(false).WorkloadPriorityClass("test").Obj(),
+			newJob:             testingutil.MakeJob("job", "default").Suspend(false).Obj(),
+			wantValidationErrs: apivalidation.ValidateImmutableField("", "test", workloadPriorityClassNamePath),
+		},
+		{
+			name:   "set priority-class when job suspend",
+			oldJob: testingutil.MakeJob("job", "default").Suspend(true).Obj(),
+			newJob: testingutil.MakeJob("job", "default").Suspend(true).WorkloadPriorityClass("test").Obj(),
+		},
+		{
+			name:   "update priority-class when job suspend",
+			oldJob: testingutil.MakeJob("job", "default").Suspend(true).WorkloadPriorityClass("test").Obj(),
+			newJob: testingutil.MakeJob("job", "default").Suspend(true).WorkloadPriorityClass("new-test").Obj(),
+		},
+		{
+			name:               "delete priority-class when job suspend",
+			oldJob:             testingutil.MakeJob("job", "default").Suspend(true).WorkloadPriorityClass("test").Obj(),
+			newJob:             testingutil.MakeJob("job", "default").Suspend(true).Obj(),
+			wantValidationErrs: apivalidation.ValidateImmutableField("", "test", workloadPriorityClassNamePath),
 		},
 		{
 			name: "immutable prebuilt workload ",
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
index ea4e28c8f..9ec4fadb2 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook_test.go
@@ -674,6 +674,25 @@ func TestValidateUpdate(t *testing.T) {
 				Label(constants.WorkloadPriorityClassLabel, "new-test").
 				Obj(),
 		},
+		"set priority class when replicas ready": {
+			oldObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			wantErr: field.ErrorList{
+				&field.Error{
+					Type:  field.ErrorTypeInvalid,
+					Field: "metadata.labels[kueue.x-k8s.io/priority-class]",
+				},
+			}.ToAggregate(),
+		},
 		"change priority class when replicas ready": {
 			oldObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
 				LeaderTemplate(corev1.PodTemplateSpec{}).
@@ -687,6 +706,64 @@ func TestValidateUpdate(t *testing.T) {
 				Label(constants.WorkloadPriorityClassLabel, "new-test").
 				ReadyReplicas(int32(1)).
 				Obj(),
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"delete priority class when replicas ready": {
+			oldObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			wantErr: field.ErrorList{
+				&field.Error{
+					Type:  field.ErrorTypeInvalid,
+					Field: "metadata.labels[kueue.x-k8s.io/priority-class]",
+				},
+			}.ToAggregate(),
+		},
+		"set priority class when replicas not ready": {
+			oldObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				Obj(),
+			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				Obj(),
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"change priority class when replicas not ready": {
+			oldObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				Obj(),
+			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "new-test").
+				ReadyReplicas(int32(1)).
+				Obj(),
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"delete priority class when replicas not ready": {
+			oldObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				Label(constants.WorkloadPriorityClassLabel, "test").
+				Obj(),
+			newObj: testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "").
+				LeaderTemplate(corev1.PodTemplateSpec{}).
+				Queue("test-queue").
+				ReadyReplicas(int32(1)).
+				Obj(),
 			wantErr: field.ErrorList{
 				&field.Error{
 					Type:  field.ErrorTypeInvalid,
diff --git a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
index c8d58b6e0..a11363518 100644
--- a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
+++ b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
@@ -376,6 +376,35 @@ func TestValidateUpdate(t *testing.T) {
 				},
 			},
 		},
+		"set in priority class label when replicas ready": {
+			oldObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel: "queue1",
+					},
+				},
+				Status: appsv1.StatefulSetStatus{
+					ReadyReplicas: int32(1),
+				},
+			},
+			newObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel:                 "queue1",
+						constants.WorkloadPriorityClassLabel: "priority2",
+					},
+				},
+				Status: appsv1.StatefulSetStatus{
+					ReadyReplicas: int32(1),
+				},
+			},
+			wantErr: field.ErrorList{
+				&field.Error{
+					Type:  field.ErrorTypeInvalid,
+					Field: priorityClassNameLabelPath.String(),
+				},
+			}.ToAggregate(),
+		},
 		"change in priority class label when replicas ready": {
 			oldObj: &appsv1.StatefulSet{
 				ObjectMeta: metav1.ObjectMeta{
@@ -399,6 +428,90 @@ func TestValidateUpdate(t *testing.T) {
 					ReadyReplicas: int32(1),
 				},
 			},
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"delete in priority class label when replicas ready": {
+			oldObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel:                 "queue1",
+						constants.WorkloadPriorityClassLabel: "priority1",
+					},
+				},
+				Status: appsv1.StatefulSetStatus{
+					ReadyReplicas: int32(1),
+				},
+			},
+			newObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel: "queue1",
+					},
+				},
+				Status: appsv1.StatefulSetStatus{
+					ReadyReplicas: int32(1),
+				},
+			},
+			wantErr: field.ErrorList{
+				&field.Error{
+					Type:  field.ErrorTypeInvalid,
+					Field: priorityClassNameLabelPath.String(),
+				},
+			}.ToAggregate(),
+		},
+		"set in priority class label when replicas not ready": {
+			oldObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel: "queue1",
+					},
+				},
+			},
+			newObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel:                 "queue1",
+						constants.WorkloadPriorityClassLabel: "priority2",
+					},
+				},
+			},
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"change in priority class label when replicas not ready": {
+			oldObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel:                 "queue1",
+						constants.WorkloadPriorityClassLabel: "priority1",
+					},
+				},
+			},
+			newObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel:                 "queue1",
+						constants.WorkloadPriorityClassLabel: "priority2",
+					},
+				},
+			},
+			wantErr: field.ErrorList{}.ToAggregate(),
+		},
+		"delete in priority class label when replicas not ready": {
+			oldObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel:                 "queue1",
+						constants.WorkloadPriorityClassLabel: "priority1",
+					},
+				},
+			},
+			newObj: &appsv1.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						constants.QueueLabel: "queue1",
+					},
+				},
+			},
 			wantErr: field.ErrorList{
 				&field.Error{
 					Type:  field.ErrorTypeInvalid,
diff --git a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
index fcb78a34a..d50563d40 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -296,53 +296,6 @@ var _ = ginkgo.Describe("Job controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 		}, util.Timeout, util.Interval).Should(gomega.Succeed())
 	})
 
-	ginkgo.It("Should sync workload priority when job priority label changes", func() {
-		priorityClass := utiltestingapi.MakeWorkloadPriorityClass(priorityClassName).
-			PriorityValue(int32(priorityValue)).Obj()
-		util.MustCreate(ctx, k8sClient, priorityClass)
-		ginkgo.DeferCleanup(func() {
-			gomega.Expect(k8sClient.Delete(ctx, priorityClass)).To(gomega.Succeed())
-		})
-
-		highPriorityClass := utiltestingapi.MakeWorkloadPriorityClass(highPriorityClassName).
-			PriorityValue(int32(highPriorityValue)).Obj()
-		util.MustCreate(ctx, k8sClient, highPriorityClass)
-		ginkgo.DeferCleanup(func() {
-			gomega.Expect(k8sClient.Delete(ctx, highPriorityClass)).To(gomega.Succeed())
-		})
-
-		ginkgo.By("creating job with priority")
-		job := testingjob.MakeJob(jobName, ns.Name).
-			WorkloadPriorityClass(priorityClassName).
-			Obj()
-		util.MustCreate(ctx, k8sClient, job)
-		lookupKey := types.NamespacedName{Name: jobName, Namespace: ns.Name}
-		createdJob := &batchv1.Job{}
-		gomega.Eventually(func(g gomega.Gomega) {
-			g.Expect(k8sClient.Get(ctx, lookupKey, createdJob)).Should(gomega.Succeed())
-			g.Expect(createdJob.Spec.Suspend).Should(gomega.Equal(ptr.To(true)))
-		}, util.Timeout, util.Interval).Should(gomega.Succeed())
-
-		ginkgo.By("checking the workload is created with priority, priorityName")
-		createdWorkload := &kueue.Workload{}
-		wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: ns.Name}
-		gomega.Eventually(func(g gomega.Gomega) {
-			g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
-		}, util.Timeout, util.Interval).Should(gomega.Succeed())
-		gomega.Expect(createdWorkload.Spec.PriorityClassName).Should(gomega.Equal(priorityClassName))
-		gomega.Expect(*createdWorkload.Spec.Priority).Should(gomega.Equal(int32(priorityValue)))
-
-		ginkgo.By("checking the workload priority is updated when the job priority label changes")
-		gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob)).Should(gomega.Succeed())
-		createdJob.Labels[constants.WorkloadPriorityClassLabel] = highPriorityClassName
-		gomega.Expect(k8sClient.Update(ctx, createdJob)).Should(gomega.Succeed())
-		gomega.Eventually(func(g gomega.Gomega) {
-			g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
-			g.Expect(createdWorkload.Spec.PriorityClassName).Should(gomega.Equal(highPriorityClassName))
-			g.Expect(*createdWorkload.Spec.Priority).Should(gomega.Equal(int32(highPriorityValue)))
-		}, util.Timeout, util.Interval).Should(gomega.Succeed())
-	})
-
 	ginkgo.It("Should reconcile job when queueName set by label", func() {
 		ginkgo.By("checking the workload is created with correct queue name assigned")
 		var jobQueueName kueue.LocalQueueName = "test-queue"
@@ -1380,6 +1333,216 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 		util.ExpectObjectToBeDeleted(ctx, k8sClient, spotUntaintedFlavor, true)
 	})
 
+	ginkgo.When("Use a WorkloadPriorityClass", func() {
+		var (
+			lowWorkloadPriorityClass  *kueue.WorkloadPriorityClass
+			highWorkloadPriorityClass *kueue.WorkloadPriorityClass
+		)
+
+		ginkgo.BeforeEach(func() {
+			lowWorkloadPriorityClass = utiltestingapi.MakeWorkloadPriorityClass("low").PriorityValue(100).Obj()
+			util.MustCreate(ctx, k8sClient, lowWorkloadPriorityClass)
+
+			highWorkloadPriorityClass = utiltestingapi.MakeWorkloadPriorityClass("high").PriorityValue(1000).Obj()
+			util.MustCreate(ctx, k8sClient, highWorkloadPriorityClass)
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(k8sClient.Delete(ctx, lowWorkloadPriorityClass)).To(gomega.Succeed())
+			gomega.Expect(k8sClient.Delete(ctx, highWorkloadPriorityClass)).To(gomega.Succeed())
+		})
+
+		ginkgo.It("running workload priority is decreased causing a pending workload to be scheduled and preempt the running one", func() {
+			job1 := testingjob.MakeJob(jobName+"1", ns.Name).
+				WorkloadPriorityClass(highWorkloadPriorityClass.Name).
+				Queue(kueue.LocalQueueName(devLocalQ.Name)).
+				Request(corev1.ResourceCPU, "5").
+				Obj()
+			ginkgo.By("creating the first Job with high priority", func() {
+				util.MustCreate(ctx, k8sClient, job1)
+			})
+
+			createdWorkload1 := &kueue.Workload{}
+			wlLookupKey1 := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job1.Name, job1.UID), Namespace: ns.Name}
+
+			ginkgo.By("checking that the first workload is created with a priority and admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).Should(gomega.Succeed())
+					g.Expect(createdWorkload1.Spec.PriorityClassName).Should(gomega.Equal(highWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload1.Spec.Priority).Should(gomega.Equal(highWorkloadPriorityClass.Value))
+					g.Expect(createdWorkload1.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			job2 := testingjob.MakeJob(jobName+"2", ns.Name).
+				WorkloadPriorityClass(highWorkloadPriorityClass.Name).
+				Queue(kueue.LocalQueueName(devLocalQ.Name)).
+				Request(corev1.ResourceCPU, "5").
+				Obj()
+			ginkgo.By("creating the second Job with high priority", func() {
+				util.MustCreate(ctx, k8sClient, job2)
+			})
+
+			createdWorkload2 := &kueue.Workload{}
+			wlLookupKey2 := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job2.Name, job2.UID), Namespace: ns.Name}
+
+			ginkgo.By("checking that the second workload is created with a priority and admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).Should(gomega.Succeed())
+					g.Expect(createdWorkload2.Spec.PriorityClassName).Should(gomega.Equal(highWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload2.Spec.Priority).Should(gomega.Equal(highWorkloadPriorityClass.Value))
+					g.Expect(createdWorkload2.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			job3 := testingjob.MakeJob(jobName+"3", ns.Name).
+				WorkloadPriorityClass(highWorkloadPriorityClass.Name).
+				Queue(kueue.LocalQueueName(devLocalQ.Name)).
+				Request(corev1.ResourceCPU, "5").
+				Obj()
+			ginkgo.By("creating the third Job with high priority", func() {
+				util.MustCreate(ctx, k8sClient, job3)
+			})
+
+			createdWorkload3 := &kueue.Workload{}
+			wlLookupKey3 := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job3.Name, job3.UID), Namespace: ns.Name}
+
+			ginkgo.By("checking that the third workload is created with a priority and not admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey3, createdWorkload3)).Should(gomega.Succeed())
+					g.Expect(createdWorkload3.Spec.PriorityClassName).Should(gomega.Equal(highWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload3.Spec.Priority).Should(gomega.Equal(highWorkloadPriorityClass.Value))
+					g.Expect(createdWorkload3.Status.Conditions).Should(testing.HaveConditionStatusFalseAndReason(kueue.WorkloadQuotaReserved, "Pending"))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("updating the second workload’s WorkloadPriorityClass to low", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdJob := &batchv1.Job{}
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(job2), createdJob)).Should(gomega.Succeed())
+					createdJob.Labels[constants.WorkloadPriorityClassLabel] = lowWorkloadPriorityClass.Name
+					g.Expect(k8sClient.Update(ctx, createdJob)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("checking that the second workload’s WorkloadPriorityClass is updated when the job’s priority label changes", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).Should(gomega.Succeed())
+					g.Expect(createdWorkload2.Spec.PriorityClassName).Should(gomega.Equal(lowWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload2.Spec.Priority).Should(gomega.Equal(lowWorkloadPriorityClass.Value))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("checking that the third workload is admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey3, createdWorkload3)).Should(gomega.Succeed())
+					g.Expect(createdWorkload3.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("checking that the second workload is evicted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).Should(gomega.Succeed())
+					g.Expect(createdWorkload2.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadEvicted))
+					g.Expect(createdWorkload2.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadPreempted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+
+		ginkgo.It("running workload priority is decreased, but still above the pending workload, and so the workload continues running undistrurbed", func() {
+			job1 := testingjob.MakeJob(jobName+"1", ns.Name).
+				WorkloadPriorityClass(highWorkloadPriorityClass.Name).
+				Queue(kueue.LocalQueueName(devLocalQ.Name)).
+				Request(corev1.ResourceCPU, "5").
+				Obj()
+			ginkgo.By("creating the first Job with high priority", func() {
+				util.MustCreate(ctx, k8sClient, job1)
+			})
+
+			createdWorkload1 := &kueue.Workload{}
+			wlLookupKey1 := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job1.Name, job1.UID), Namespace: ns.Name}
+
+			ginkgo.By("checking that the first workload is created with a priority and admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).Should(gomega.Succeed())
+					g.Expect(createdWorkload1.Spec.PriorityClassName).Should(gomega.Equal(highWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload1.Spec.Priority).Should(gomega.Equal(highWorkloadPriorityClass.Value))
+					g.Expect(createdWorkload1.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			job2 := testingjob.MakeJob(jobName+"2", ns.Name).
+				WorkloadPriorityClass(highWorkloadPriorityClass.Name).
+				Queue(kueue.LocalQueueName(devLocalQ.Name)).
+				Request(corev1.ResourceCPU, "5").
+				Obj()
+			ginkgo.By("creating the second Job with high priority", func() {
+				util.MustCreate(ctx, k8sClient, job2)
+			})
+
+			createdWorkload2 := &kueue.Workload{}
+			wlLookupKey2 := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job2.Name, job2.UID), Namespace: ns.Name}
+
+			ginkgo.By("checking that the second workload is created with a priority and admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).Should(gomega.Succeed())
+					g.Expect(createdWorkload2.Spec.PriorityClassName).Should(gomega.Equal(highWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload2.Spec.Priority).Should(gomega.Equal(highWorkloadPriorityClass.Value))
+					g.Expect(createdWorkload2.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			job3 := testingjob.MakeJob(jobName+"3", ns.Name).
+				WorkloadPriorityClass(lowWorkloadPriorityClass.Name).
+				Queue(kueue.LocalQueueName(devLocalQ.Name)).
+				Request(corev1.ResourceCPU, "5").
+				Obj()
+			ginkgo.By("creating the third Job with low priority", func() {
+				util.MustCreate(ctx, k8sClient, job3)
+			})
+
+			createdWorkload3 := &kueue.Workload{}
+			wlLookupKey3 := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job3.Name, job3.UID), Namespace: ns.Name}
+
+			ginkgo.By("checking that the third workload is created with a priority and not admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey3, createdWorkload3)).Should(gomega.Succeed())
+					g.Expect(createdWorkload3.Spec.PriorityClassName).Should(gomega.Equal(lowWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload3.Spec.Priority).Should(gomega.Equal(lowWorkloadPriorityClass.Value))
+					g.Expect(createdWorkload3.Status.Conditions).Should(testing.HaveConditionStatusFalseAndReason(kueue.WorkloadQuotaReserved, "Pending"))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("updating the second workload’s WorkloadPriorityClass to low", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdJob := &batchv1.Job{}
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(job2), createdJob)).Should(gomega.Succeed())
+					createdJob.Labels[constants.WorkloadPriorityClassLabel] = lowWorkloadPriorityClass.Name
+					g.Expect(k8sClient.Update(ctx, createdJob)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("checking that the second workload’s WorkloadPriorityClass is updated when the job’s priority label changes", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).Should(gomega.Succeed())
+					g.Expect(createdWorkload2.Spec.PriorityClassName).Should(gomega.Equal(lowWorkloadPriorityClass.Name))
+					g.Expect(*createdWorkload2.Spec.Priority).Should(gomega.Equal(lowWorkloadPriorityClass.Value))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("checking that the third workload is not admitted and the second workload is not evicted", func() {
+				gomega.Consistently(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).Should(gomega.Succeed())
+					g.Expect(createdWorkload2.Status.Conditions).Should(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+					g.Expect(createdWorkload2.Status.Conditions).ShouldNot(testing.HaveConditionStatusTrue(kueue.WorkloadEvicted))
+					g.Expect(createdWorkload2.Status.Conditions).ShouldNot(testing.HaveConditionStatusTrue(kueue.WorkloadPreempted))
+					g.Expect(k8sClient.Get(ctx, wlLookupKey3, createdWorkload3)).Should(gomega.Succeed())
+					g.Expect(createdWorkload3.Status.Conditions).Should(testing.HaveConditionStatusFalseAndReason(kueue.WorkloadQuotaReserved, "Pending"))
+				}, util.ConsistentDuration, util.ConsistentDuration).Should(gomega.Succeed())
+			})
+		})
+	})
+
 	ginkgo.It("Should schedule jobs as they fit in their ClusterQueue", framework.SlowSpec, func() {
 		ginkgo.By("checking the first prod job starts")
 		prodJob1 := testingjob.MakeJob("prod-job1", ns.Name).Queue(kueue.LocalQueueName(prodLocalQ.Name)).Request(corev1.ResourceCPU, "2").Obj()
diff --git a/test/integration/singlecluster/webhook/core/workload_test.go b/test/integration/singlecluster/webhook/core/workload_test.go
index 6ba2543e9..ac4395017 100644
--- a/test/integration/singlecluster/webhook/core/workload_test.go
+++ b/test/integration/singlecluster/webhook/core/workload_test.go
@@ -802,6 +802,178 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				testing.BeInvalidError(),
 			),
+			ginkgo.Entry("can set workload priority class when QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = constants.WorkloadPriorityClassSource
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can't set workload priority class when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
+				},
+				true,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = constants.WorkloadPriorityClassSource
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				testing.BeInvalidError(),
+			),
+			ginkgo.Entry("can update workload priority class when QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.WorkloadPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can update workload priority class when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.WorkloadPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				true,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can delete workload priority class when QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.WorkloadPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = ""
+					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.Priority = nil
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can't delete workload priority class when quota reserved when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.WorkloadPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				true,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = ""
+					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.Priority = nil
+				},
+				testing.BeInvalidError(),
+			),
+			ginkgo.Entry("can set pod priority class QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = constants.PodPriorityClassSource
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can't set pod priority class when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
+				},
+				true,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = constants.PodPriorityClassSource
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				testing.BeInvalidError(),
+			),
+			ginkgo.Entry("can update pod priority class when QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.PodPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can't update pod priority class when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.PodPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				true,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.Priority = ptr.To[int32](100)
+				},
+				testing.BeInvalidError(),
+			),
+			ginkgo.Entry("can delete pod priority class when QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.PodPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = ""
+					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.Priority = nil
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can't delete pod priority class when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassSource(constants.PodPriorityClassSource).
+						PriorityClass("high").
+						Priority(1000).
+						Obj()
+				},
+				true,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.PriorityClassSource = ""
+					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.Priority = nil
+				},
+				testing.BeInvalidError(),
+			),
 		)
 
 		ginkgo.It("Should forbid the change of spec.queueName of an admitted workload", func() {