Repository: kubernetes-sigs/kueue

Issue:
v1beta2: change the API for Workload's spec.priorityClassSource
<!-- Please only use this template for submitting clean up requests -->

**What would you like to be cleaned**:

Two possibilities are disucssed:
1. `priorityClassSource` field as *string.
2. `priorityClassRef`


**Why is this needed**:

Part of https://github.com/kubernetes-sigs/kueue/issues/7113

The field is causing troubles when conversion webhooks are enabled: https://github.com/kubernetes-sigs/kueue/pull/7318#discussion_r2450893646

Also, ptr (optional) would better reflect the intention since the field may not necessarily be specified.


Discussion:
@tenzen-y: /retitle v1beta2: change the API for Workload's spec.priorityClassSource

The priorityClassSource is located in spec.@tenzen-y: Let me expand (2) here a bit more
The original motivation is to align with the Kubernetes API recommendation: https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md#multiple-resource-reference

We can consider the following new structure:

```yaml
spec:
  priority: 123456
  priorityClassRef:
    group: kueue.x-k8s.io | scheduling.k8s.io
    kind: PriorityClass | WorkloadPriorityClass
    name: high
```

If we select the above objectRef pattern, we will remove `.spec.priorityClassName` since the name will be included in the `.spec.priorityClassRef.name`.

Initially posted in https://github.com/kubernetes-sigs/kueue/pull/7318#issuecomment-3432748098
@mimowo: +1, the only consideration I think is that when we convert in webhooks from v1beta1, then we will not be able to populate `priorityClassRef.name` directly based on the old object. 

Some options I can see:
1. keep the `name ` as ptr for this reason
2. don't use ptr, but just add comment / docs disclaimer that for workloads created in v1beta1 it may be empty string "" (which may not be accurate)
3. lookup the priority class name

I think we can safely exclude (3.). I think (2.) makes sense actually because this is a problem only in transition period.
@tenzen-y: > I think we can safely exclude (3.). I think (2.) makes sense actually because this is a problem only in transition period.

LGTM for opt (2).@mbobrovskyi: /assign

Solution patch:
diff --git a/apis/kueue/v1beta1/workload_conversion.go b/apis/kueue/v1beta1/workload_conversion.go
index 55d407ec5..729b7f863 100644
--- a/apis/kueue/v1beta1/workload_conversion.go
+++ b/apis/kueue/v1beta1/workload_conversion.go
@@ -45,3 +45,27 @@ func Convert_v1beta1_WorkloadStatus_To_v1beta2_WorkloadStatus(in *WorkloadStatus
 	out.AccumulatedPastExecutionTimeSeconds = in.AccumulatedPastExexcutionTimeSeconds
 	return autoConvert_v1beta1_WorkloadStatus_To_v1beta2_WorkloadStatus(in, out, s)
 }
+
+func Convert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec(in *WorkloadSpec, out *v1beta2.WorkloadSpec, s conversionapi.Scope) error {
+	switch in.PriorityClassSource {
+	case WorkloadPriorityClassSource:
+		out.PriorityClassRef = v1beta2.NewWorkloadPriorityClassRef(in.PriorityClassName)
+	case PodPriorityClassSource:
+		out.PriorityClassRef = v1beta2.NewPodPriorityClassRef(in.PriorityClassName)
+	}
+	return autoConvert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec(in, out, s)
+}
+
+func Convert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec(in *v1beta2.WorkloadSpec, out *WorkloadSpec, s conversionapi.Scope) error {
+	if in.PriorityClassRef != nil {
+		switch {
+		case in.PriorityClassRef.Group == v1beta2.WorkloadPriorityClassGroup && in.PriorityClassRef.Kind == v1beta2.WorkloadPriorityClassKind:
+			out.PriorityClassSource = WorkloadPriorityClassSource
+			out.PriorityClassName = in.PriorityClassRef.Name
+		case in.PriorityClassRef.Group == v1beta2.PodPriorityClassGroup && in.PriorityClassRef.Kind == v1beta2.PodPriorityClassKind:
+			out.PriorityClassSource = PodPriorityClassSource
+			out.PriorityClassName = in.PriorityClassRef.Name
+		}
+	}
+	return autoConvert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec(in, out, s)
+}
diff --git a/apis/kueue/v1beta1/workload_types.go b/apis/kueue/v1beta1/workload_types.go
index 1c1d44696..86ab44baa 100644
--- a/apis/kueue/v1beta1/workload_types.go
+++ b/apis/kueue/v1beta1/workload_types.go
@@ -23,6 +23,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const (
+	WorkloadPriorityClassSource = "kueue.x-k8s.io/workloadpriorityclass"
+	PodPriorityClassSource      = "scheduling.k8s.io/priorityclass"
+)
+
 // WorkloadSpec defines the desired state of Workload
 // +kubebuilder:validation:XValidation:rule="has(self.priorityClassName) ? has(self.priority) : true", message="priority should not be nil when priorityClassName is set"
 type WorkloadSpec struct {
diff --git a/apis/kueue/v1beta1/zz_generated.conversion.go b/apis/kueue/v1beta1/zz_generated.conversion.go
index a95e8f219..3ae4e0220 100644
--- a/apis/kueue/v1beta1/zz_generated.conversion.go
+++ b/apis/kueue/v1beta1/zz_generated.conversion.go
@@ -738,16 +738,6 @@ func RegisterConversions(s *runtime.Scheme) error {
 	}); err != nil {
 		return err
 	}
-	if err := s.AddGeneratedConversionFunc((*WorkloadSpec)(nil), (*v1beta2.WorkloadSpec)(nil), func(a, b interface{}, scope conversion.Scope) error {
-		return Convert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec(a.(*WorkloadSpec), b.(*v1beta2.WorkloadSpec), scope)
-	}); err != nil {
-		return err
-	}
-	if err := s.AddGeneratedConversionFunc((*v1beta2.WorkloadSpec)(nil), (*WorkloadSpec)(nil), func(a, b interface{}, scope conversion.Scope) error {
-		return Convert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec(a.(*v1beta2.WorkloadSpec), b.(*WorkloadSpec), scope)
-	}); err != nil {
-		return err
-	}
 	if err := s.AddConversionFunc((*AdmissionCheckSpec)(nil), (*v1beta2.AdmissionCheckSpec)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1beta1_AdmissionCheckSpec_To_v1beta2_AdmissionCheckSpec(a.(*AdmissionCheckSpec), b.(*v1beta2.AdmissionCheckSpec), scope)
 	}); err != nil {
@@ -768,6 +758,11 @@ func RegisterConversions(s *runtime.Scheme) error {
 	}); err != nil {
 		return err
 	}
+	if err := s.AddConversionFunc((*WorkloadSpec)(nil), (*v1beta2.WorkloadSpec)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec(a.(*WorkloadSpec), b.(*v1beta2.WorkloadSpec), scope)
+	}); err != nil {
+		return err
+	}
 	if err := s.AddConversionFunc((*WorkloadStatus)(nil), (*v1beta2.WorkloadStatus)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1beta1_WorkloadStatus_To_v1beta2_WorkloadStatus(a.(*WorkloadStatus), b.(*v1beta2.WorkloadStatus), scope)
 	}); err != nil {
@@ -783,6 +778,11 @@ func RegisterConversions(s *runtime.Scheme) error {
 	}); err != nil {
 		return err
 	}
+	if err := s.AddConversionFunc((*v1beta2.WorkloadSpec)(nil), (*WorkloadSpec)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec(a.(*v1beta2.WorkloadSpec), b.(*WorkloadSpec), scope)
+	}); err != nil {
+		return err
+	}
 	if err := s.AddConversionFunc((*v1beta2.WorkloadStatus)(nil), (*WorkloadStatus)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1beta2_WorkloadStatus_To_v1beta1_WorkloadStatus(a.(*v1beta2.WorkloadStatus), b.(*WorkloadStatus), scope)
 	}); err != nil {
@@ -2616,35 +2616,24 @@ func Convert_v1beta2_WorkloadSchedulingStatsEviction_To_v1beta1_WorkloadScheduli
 func autoConvert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec(in *WorkloadSpec, out *v1beta2.WorkloadSpec, s conversion.Scope) error {
 	out.PodSets = *(*[]v1beta2.PodSet)(unsafe.Pointer(&in.PodSets))
 	out.QueueName = v1beta2.LocalQueueName(in.QueueName)
-	out.PriorityClassName = in.PriorityClassName
+	// WARNING: in.PriorityClassName requires manual conversion: does not exist in peer-type
 	out.Priority = (*int32)(unsafe.Pointer(in.Priority))
-	out.PriorityClassSource = in.PriorityClassSource
+	// WARNING: in.PriorityClassSource requires manual conversion: does not exist in peer-type
 	out.Active = (*bool)(unsafe.Pointer(in.Active))
 	out.MaximumExecutionTimeSeconds = (*int32)(unsafe.Pointer(in.MaximumExecutionTimeSeconds))
 	return nil
 }
 
-// Convert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec is an autogenerated conversion function.
-func Convert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec(in *WorkloadSpec, out *v1beta2.WorkloadSpec, s conversion.Scope) error {
-	return autoConvert_v1beta1_WorkloadSpec_To_v1beta2_WorkloadSpec(in, out, s)
-}
-
 func autoConvert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec(in *v1beta2.WorkloadSpec, out *WorkloadSpec, s conversion.Scope) error {
 	out.PodSets = *(*[]PodSet)(unsafe.Pointer(&in.PodSets))
 	out.QueueName = LocalQueueName(in.QueueName)
-	out.PriorityClassName = in.PriorityClassName
+	// WARNING: in.PriorityClassRef requires manual conversion: does not exist in peer-type
 	out.Priority = (*int32)(unsafe.Pointer(in.Priority))
-	out.PriorityClassSource = in.PriorityClassSource
 	out.Active = (*bool)(unsafe.Pointer(in.Active))
 	out.MaximumExecutionTimeSeconds = (*int32)(unsafe.Pointer(in.MaximumExecutionTimeSeconds))
 	return nil
 }
 
-// Convert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec is an autogenerated conversion function.
-func Convert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec(in *v1beta2.WorkloadSpec, out *WorkloadSpec, s conversion.Scope) error {
-	return autoConvert_v1beta2_WorkloadSpec_To_v1beta1_WorkloadSpec(in, out, s)
-}
-
 func autoConvert_v1beta1_WorkloadStatus_To_v1beta2_WorkloadStatus(in *WorkloadStatus, out *v1beta2.WorkloadStatus, s conversion.Scope) error {
 	out.Conditions = *(*[]v1.Condition)(unsafe.Pointer(&in.Conditions))
 	out.Admission = (*v1beta2.Admission)(unsafe.Pointer(in.Admission))
diff --git a/apis/kueue/v1beta2/workload_types.go b/apis/kueue/v1beta2/workload_types.go
index 556e9151b..825eea6a6 100644
--- a/apis/kueue/v1beta2/workload_types.go
+++ b/apis/kueue/v1beta2/workload_types.go
@@ -24,7 +24,7 @@ import (
 )
 
 // WorkloadSpec defines the desired state of Workload
-// +kubebuilder:validation:XValidation:rule="has(self.priorityClassName) ? has(self.priority) : true", message="priority should not be nil when priorityClassName is set"
+// +kubebuilder:validation:XValidation:rule="!has(self.priorityClassRef) || has(self.priority)", message="priority should not be nil when priorityClassRef is set"
 type WorkloadSpec struct {
 	// podSets is a list of sets of homogeneous pods, each described by a Pod spec
 	// and a count.
@@ -43,35 +43,19 @@ type WorkloadSpec struct {
 	// +optional
 	QueueName LocalQueueName `json:"queueName,omitempty"`
 
-	// priorityClassName is the name of the PriorityClass the Workload is associated with.
-	// If specified, indicates the workload's priority.
-	// "system-node-critical" and "system-cluster-critical" are two special
-	// keywords which indicate the highest priorities with the former being
-	// the highest priority. Any other name must be defined by creating a
-	// PriorityClass object with that name. If not specified, the workload
-	// priority will be default or zero if there is no default.
-	// +kubebuilder:validation:MaxLength=253
-	// +kubebuilder:validation:MinLength=1
-	// +kubebuilder:validation:Pattern="^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
+	// priorityClassRef references a PriorityClass object that defines the workload's priority.
+	//
 	// +optional
-	PriorityClassName string `json:"priorityClassName,omitempty"`
+	PriorityClassRef *PriorityClassRef `json:"priorityClassRef,omitempty"`
 
 	// priority determines the order of access to the resources managed by the
 	// ClusterQueue where the workload is queued.
-	// The priority value is populated from PriorityClassName.
+	// The priority value is populated from the referenced PriorityClass (via priorityClassRef).
 	// The higher the value, the higher the priority.
-	// If priorityClassName is specified, priority must not be null.
+	// If priorityClassRef is specified, priority must not be null.
 	// +optional
 	Priority *int32 `json:"priority,omitempty"`
 
-	// priorityClassSource determines whether the priorityClass field refers to a pod PriorityClass or kueue.x-k8s.io/workloadpriorityclass.
-	// Workload's PriorityClass can accept the name of a pod priorityClass or a workloadPriorityClass.
-	// When using pod PriorityClass, a priorityClassSource field has the scheduling.k8s.io/priorityclass value.
-	// +kubebuilder:default=""
-	// +kubebuilder:validation:Enum=kueue.x-k8s.io/workloadpriorityclass;scheduling.k8s.io/priorityclass;""
-	// +optional
-	PriorityClassSource string `json:"priorityClassSource,omitempty"`
-
 	// active determines if a workload can be admitted into a queue.
 	// Changing active from true to false will evict any running workloads.
 	// Possible values are:
@@ -93,6 +77,81 @@ type WorkloadSpec struct {
 	MaximumExecutionTimeSeconds *int32 `json:"maximumExecutionTimeSeconds,omitempty"`
 }
 
+// PriorityClassGroup indicates the API group of the PriorityClass object.
+//
+// +enum
+// +kubebuilder:validation:Enum=kueue.x-k8s.io;scheduling.k8s.io
+type PriorityClassGroup string
+
+const (
+	// PodPriorityClassGroup represents the core Kubernetes scheduling group.
+	PodPriorityClassGroup PriorityClassGroup = "scheduling.k8s.io"
+
+	// WorkloadPriorityClassGroup represents the Kueue-specific priority group.
+	WorkloadPriorityClassGroup PriorityClassGroup = "kueue.x-k8s.io"
+)
+
+// PriorityClassKind is the kind of the PriorityClass object.
+//
+// +enum
+// +kubebuilder:validation:Enum=WorkloadPriorityClass;PriorityClass
+type PriorityClassKind string
+
+const (
+	// PodPriorityClassKind represents the core Kubernetes PriorityClass kind.
+	PodPriorityClassKind PriorityClassKind = "PriorityClass"
+
+	// WorkloadPriorityClassKind represents the Kueue WorkloadPriorityClass kind.
+	WorkloadPriorityClassKind PriorityClassKind = "WorkloadPriorityClass"
+)
+
+// NewPriorityClassRef creates a new PriorityClassRef with the specified group, kind, and name.
+func NewPriorityClassRef(group PriorityClassGroup, kind PriorityClassKind, name string) *PriorityClassRef {
+	return &PriorityClassRef{Group: group, Kind: kind, Name: name}
+}
+
+// NewWorkloadPriorityClassRef creates a PriorityClassRef for a WorkloadPriorityClass.
+func NewWorkloadPriorityClassRef(name string) *PriorityClassRef {
+	return NewPriorityClassRef(WorkloadPriorityClassGroup, WorkloadPriorityClassKind, name)
+}
+
+// NewPodPriorityClassRef creates a PriorityClassRef for a core Kubernetes PriorityClass.
+func NewPodPriorityClassRef(name string) *PriorityClassRef {
+	return NewPriorityClassRef(PodPriorityClassGroup, PodPriorityClassKind, name)
+}
+
+// PriorityClassRef references a PriorityClass in a specific API group.
+//
+// +kubebuilder:validation:XValidation:rule="(self.group == 'scheduling.k8s.io') ? self.kind == 'PriorityClass' : true", message="only the PriorityClass kind is allowed for the scheduling.k8s.io group"
+// +kubebuilder:validation:XValidation:rule="(self.group == 'kueue.x-k8s.io') ? self.kind == 'WorkloadPriorityClass' : true", message="only the WorkloadPriorityClass kind is allowed for the kueue.x-k8s.io group"
+type PriorityClassRef struct {
+	// group is the API group of the PriorityClass object.
+	// Use "kueue.x-k8s.io" for WorkloadPriorityClass.
+	// Use "scheduling.k8s.io" for Pod PriorityClass.
+	//
+	// +required
+	Group PriorityClassGroup `json:"group,omitempty"`
+
+	// kind is the kind of the PriorityClass object.
+	//
+	// +required
+	Kind PriorityClassKind `json:"kind,omitempty"`
+
+	// name is the name of the PriorityClass the Workload is associated with.
+	// If specified, indicates the workload's priority.
+	// "system-node-critical" and "system-cluster-critical" are two special
+	// keywords which indicate the highest priorities with the former being
+	// the highest priority. Any other name must be defined by creating a
+	// PriorityClass object with that name. If not specified, the workload
+	// priority will be default or zero if there is no default.
+	//
+	// +required
+	// +kubebuilder:validation:MaxLength=253
+	// +kubebuilder:validation:MinLength=1
+	// +kubebuilder:validation:Pattern="^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
+	Name string `json:"name,omitempty"`
+}
+
 // PodSetTopologyRequest defines the topology request for a PodSet.
 type PodSetTopologyRequest struct {
 	// required indicates the topology level required by the PodSet, as
@@ -800,8 +859,10 @@ 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') && (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(oldSelf.spec.priorityClassRef) && !has(self.spec.priorityClassRef)) || (has(oldSelf.spec.priorityClassRef) && has(self.spec.priorityClassRef)) : true",message="priorityClassRef 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.priorityClassRef) && has(self.spec.priorityClassRef)) ? oldSelf.spec.priorityClassRef.group == self.spec.priorityClassRef.group : true",message="priorityClassRef.group 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.priorityClassRef) && has(self.spec.priorityClassRef)) ? oldSelf.spec.priorityClassRef.kind == self.spec.priorityClassRef.kind : true",message="priorityClassRef.kind 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.priorityClassRef) && has(self.spec.priorityClassRef) && self.spec.priorityClassRef.group == 'scheduling.k8s.io' && self.spec.priorityClassRef.kind == 'PriorityClass') ? oldSelf.spec.priorityClassRef.name == self.spec.priorityClassRef.name : true",message="priorityClassRef.name is immutable for scheduling.k8s.io/priorityclass 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(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/zz_generated.deepcopy.go b/apis/kueue/v1beta2/zz_generated.deepcopy.go
index b641161ba..77a57cd69 100644
--- a/apis/kueue/v1beta2/zz_generated.deepcopy.go
+++ b/apis/kueue/v1beta2/zz_generated.deepcopy.go
@@ -1217,6 +1217,21 @@ func (in *PodSetUpdate) DeepCopy() *PodSetUpdate {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PriorityClassRef) DeepCopyInto(out *PriorityClassRef) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PriorityClassRef.
+func (in *PriorityClassRef) DeepCopy() *PriorityClassRef {
+	if in == nil {
+		return nil
+	}
+	out := new(PriorityClassRef)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ProvisioningRequestConfig) DeepCopyInto(out *ProvisioningRequestConfig) {
 	*out = *in
@@ -1926,6 +1941,11 @@ func (in *WorkloadSpec) DeepCopyInto(out *WorkloadSpec) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.PriorityClassRef != nil {
+		in, out := &in.PriorityClassRef, &out.PriorityClassRef
+		*out = new(PriorityClassRef)
+		**out = **in
+	}
 	if in.Priority != nil {
 		in, out := &in.Priority, &out.Priority
 		*out = new(int32)
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 437d20e29..aba263826 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
@@ -17119,35 +17119,52 @@ spec:
                   description: |-
                     priority determines the order of access to the resources managed by the
                     ClusterQueue where the workload is queued.
-                    The priority value is populated from PriorityClassName.
+                    The priority value is populated from the referenced PriorityClass (via priorityClassRef).
                     The higher the value, the higher the priority.
-                    If priorityClassName is specified, priority must not be null.
+                    If priorityClassRef is specified, priority must not be null.
                   format: int32
                   type: integer
-                priorityClassName:
-                  description: |-
-                    priorityClassName is the name of the PriorityClass the Workload is associated with.
-                    If specified, indicates the workload's priority.
-                    "system-node-critical" and "system-cluster-critical" are two special
-                    keywords which indicate the highest priorities with the former being
-                    the highest priority. Any other name must be defined by creating a
-                    PriorityClass object with that name. If not specified, the workload
-                    priority will be default or zero if there is no default.
-                  maxLength: 253
-                  minLength: 1
-                  pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
-                  type: string
-                priorityClassSource:
-                  default: ""
-                  description: |-
-                    priorityClassSource determines whether the priorityClass field refers to a pod PriorityClass or kueue.x-k8s.io/workloadpriorityclass.
-                    Workload's PriorityClass can accept the name of a pod priorityClass or a workloadPriorityClass.
-                    When using pod PriorityClass, a priorityClassSource field has the scheduling.k8s.io/priorityclass value.
-                  enum:
-                    - kueue.x-k8s.io/workloadpriorityclass
-                    - scheduling.k8s.io/priorityclass
-                    - ""
-                  type: string
+                priorityClassRef:
+                  description: priorityClassRef references a PriorityClass object that defines the workload's priority.
+                  properties:
+                    group:
+                      description: |-
+                        group is the API group of the PriorityClass object.
+                        Use "kueue.x-k8s.io" for WorkloadPriorityClass.
+                        Use "scheduling.k8s.io" for Pod PriorityClass.
+                      enum:
+                        - kueue.x-k8s.io
+                        - scheduling.k8s.io
+                      type: string
+                    kind:
+                      description: kind is the kind of the PriorityClass object.
+                      enum:
+                        - WorkloadPriorityClass
+                        - PriorityClass
+                      type: string
+                    name:
+                      description: |-
+                        name is the name of the PriorityClass the Workload is associated with.
+                        If specified, indicates the workload's priority.
+                        "system-node-critical" and "system-cluster-critical" are two special
+                        keywords which indicate the highest priorities with the former being
+                        the highest priority. Any other name must be defined by creating a
+                        PriorityClass object with that name. If not specified, the workload
+                        priority will be default or zero if there is no default.
+                      maxLength: 253
+                      minLength: 1
+                      pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+                      type: string
+                  required:
+                    - group
+                    - kind
+                    - name
+                  type: object
+                  x-kubernetes-validations:
+                    - message: only the PriorityClass kind is allowed for the scheduling.k8s.io group
+                      rule: '(self.group == ''scheduling.k8s.io'') ? self.kind == ''PriorityClass'' : true'
+                    - message: only the WorkloadPriorityClass kind is allowed for the kueue.x-k8s.io group
+                      rule: '(self.group == ''kueue.x-k8s.io'') ? self.kind == ''WorkloadPriorityClass'' : true'
                 queueName:
                   description: |-
                     queueName is the name of the LocalQueue the Workload is associated with.
@@ -17157,8 +17174,8 @@ spec:
                   type: string
               type: object
               x-kubernetes-validations:
-                - message: priority should not be nil when priorityClassName is set
-                  rule: 'has(self.priorityClassName) ? has(self.priority) : true'
+                - message: priority should not be nil when priorityClassRef is set
+                  rule: '!has(self.priorityClassRef) || has(self.priority)'
             status:
               description: status is the status of the Workload.
               properties:
@@ -17700,10 +17717,14 @@ spec:
           x-kubernetes-validations:
             - message: podSetAssignments must have the same number of podSets as the 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 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: priorityClassRef 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.priorityClassRef) && !has(self.spec.priorityClassRef)) || (has(oldSelf.spec.priorityClassRef) && has(self.spec.priorityClassRef)) : true'
+            - message: priorityClassRef.group 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.priorityClassRef) && has(self.spec.priorityClassRef)) ? oldSelf.spec.priorityClassRef.group == self.spec.priorityClassRef.group : true'
+            - message: priorityClassRef.kind 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.priorityClassRef) && has(self.spec.priorityClassRef)) ? oldSelf.spec.priorityClassRef.kind == self.spec.priorityClassRef.kind : true'
+            - message: priorityClassRef.name is immutable for scheduling.k8s.io/priorityclass 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.priorityClassRef) && has(self.spec.priorityClassRef) && self.spec.priorityClassRef.group == ''scheduling.k8s.io'' && self.spec.priorityClassRef.kind == ''PriorityClass'') ? oldSelf.spec.priorityClassRef.name == self.spec.priorityClassRef.name : 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/client-go/applyconfiguration/kueue/v1beta2/priorityclassref.go b/client-go/applyconfiguration/kueue/v1beta2/priorityclassref.go
new file mode 100644
index 000000000..6f31340a9
--- /dev/null
+++ b/client-go/applyconfiguration/kueue/v1beta2/priorityclassref.go
@@ -0,0 +1,60 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+// Code generated by applyconfiguration-gen. DO NOT EDIT.
+
+package v1beta2
+
+import (
+	kueuev1beta2 "sigs.k8s.io/kueue/apis/kueue/v1beta2"
+)
+
+// PriorityClassRefApplyConfiguration represents a declarative configuration of the PriorityClassRef type for use
+// with apply.
+type PriorityClassRefApplyConfiguration struct {
+	Group *kueuev1beta2.PriorityClassGroup `json:"group,omitempty"`
+	Kind  *kueuev1beta2.PriorityClassKind  `json:"kind,omitempty"`
+	Name  *string                          `json:"name,omitempty"`
+}
+
+// PriorityClassRefApplyConfiguration constructs a declarative configuration of the PriorityClassRef type for use with
+// apply.
+func PriorityClassRef() *PriorityClassRefApplyConfiguration {
+	return &PriorityClassRefApplyConfiguration{}
+}
+
+// WithGroup sets the Group field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the Group field is set to the value of the last call.
+func (b *PriorityClassRefApplyConfiguration) WithGroup(value kueuev1beta2.PriorityClassGroup) *PriorityClassRefApplyConfiguration {
+	b.Group = &value
+	return b
+}
+
+// WithKind sets the Kind field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the Kind field is set to the value of the last call.
+func (b *PriorityClassRefApplyConfiguration) WithKind(value kueuev1beta2.PriorityClassKind) *PriorityClassRefApplyConfiguration {
+	b.Kind = &value
+	return b
+}
+
+// WithName sets the Name field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the Name field is set to the value of the last call.
+func (b *PriorityClassRefApplyConfiguration) WithName(value string) *PriorityClassRefApplyConfiguration {
+	b.Name = &value
+	return b
+}
diff --git a/client-go/applyconfiguration/kueue/v1beta2/workloadspec.go b/client-go/applyconfiguration/kueue/v1beta2/workloadspec.go
index ac14b4706..ab7c0b9ea 100644
--- a/client-go/applyconfiguration/kueue/v1beta2/workloadspec.go
+++ b/client-go/applyconfiguration/kueue/v1beta2/workloadspec.go
@@ -24,13 +24,12 @@ import (
 // WorkloadSpecApplyConfiguration represents a declarative configuration of the WorkloadSpec type for use
 // with apply.
 type WorkloadSpecApplyConfiguration struct {
-	PodSets                     []PodSetApplyConfiguration   `json:"podSets,omitempty"`
-	QueueName                   *kueuev1beta2.LocalQueueName `json:"queueName,omitempty"`
-	PriorityClassName           *string                      `json:"priorityClassName,omitempty"`
-	Priority                    *int32                       `json:"priority,omitempty"`
-	PriorityClassSource         *string                      `json:"priorityClassSource,omitempty"`
-	Active                      *bool                        `json:"active,omitempty"`
-	MaximumExecutionTimeSeconds *int32                       `json:"maximumExecutionTimeSeconds,omitempty"`
+	PodSets                     []PodSetApplyConfiguration          `json:"podSets,omitempty"`
+	QueueName                   *kueuev1beta2.LocalQueueName        `json:"queueName,omitempty"`
+	PriorityClassRef            *PriorityClassRefApplyConfiguration `json:"priorityClassRef,omitempty"`
+	Priority                    *int32                              `json:"priority,omitempty"`
+	Active                      *bool                               `json:"active,omitempty"`
+	MaximumExecutionTimeSeconds *int32                              `json:"maximumExecutionTimeSeconds,omitempty"`
 }
 
 // WorkloadSpecApplyConfiguration constructs a declarative configuration of the WorkloadSpec type for use with
@@ -60,11 +59,11 @@ func (b *WorkloadSpecApplyConfiguration) WithQueueName(value kueuev1beta2.LocalQ
 	return b
 }
 
-// WithPriorityClassName sets the PriorityClassName field in the declarative configuration to the given value
+// WithPriorityClassRef sets the PriorityClassRef field in the declarative configuration to the given value
 // and returns the receiver, so that objects can be built by chaining "With" function invocations.
-// If called multiple times, the PriorityClassName field is set to the value of the last call.
-func (b *WorkloadSpecApplyConfiguration) WithPriorityClassName(value string) *WorkloadSpecApplyConfiguration {
-	b.PriorityClassName = &value
+// If called multiple times, the PriorityClassRef field is set to the value of the last call.
+func (b *WorkloadSpecApplyConfiguration) WithPriorityClassRef(value *PriorityClassRefApplyConfiguration) *WorkloadSpecApplyConfiguration {
+	b.PriorityClassRef = value
 	return b
 }
 
@@ -76,14 +75,6 @@ func (b *WorkloadSpecApplyConfiguration) WithPriority(value int32) *WorkloadSpec
 	return b
 }
 
-// WithPriorityClassSource sets the PriorityClassSource field in the declarative configuration to the given value
-// and returns the receiver, so that objects can be built by chaining "With" function invocations.
-// If called multiple times, the PriorityClassSource field is set to the value of the last call.
-func (b *WorkloadSpecApplyConfiguration) WithPriorityClassSource(value string) *WorkloadSpecApplyConfiguration {
-	b.PriorityClassSource = &value
-	return b
-}
-
 // WithActive sets the Active field in the declarative configuration to the given value
 // and returns the receiver, so that objects can be built by chaining "With" function invocations.
 // If called multiple times, the Active field is set to the value of the last call.
diff --git a/client-go/applyconfiguration/utils.go b/client-go/applyconfiguration/utils.go
index ad820f3af..162c4b523 100644
--- a/client-go/applyconfiguration/utils.go
+++ b/client-go/applyconfiguration/utils.go
@@ -251,6 +251,8 @@ func ForKind(kind schema.GroupVersionKind) interface{} {
 		return &kueuev1beta2.PodSetTopologyRequestApplyConfiguration{}
 	case v1beta2.SchemeGroupVersion.WithKind("PodSetUpdate"):
 		return &kueuev1beta2.PodSetUpdateApplyConfiguration{}
+	case v1beta2.SchemeGroupVersion.WithKind("PriorityClassRef"):
+		return &kueuev1beta2.PriorityClassRefApplyConfiguration{}
 	case v1beta2.SchemeGroupVersion.WithKind("ProvisioningRequestConfig"):
 		return &kueuev1beta2.ProvisioningRequestConfigApplyConfiguration{}
 	case v1beta2.SchemeGroupVersion.WithKind("ProvisioningRequestConfigSpec"):
diff --git a/cmd/importer/pod/import.go b/cmd/importer/pod/import.go
index 6cadf7a7d..8ffa694ee 100644
--- a/cmd/importer/pod/import.go
+++ b/cmd/importer/pod/import.go
@@ -80,9 +80,8 @@ func Import(ctx context.Context, c client.Client, cache *util.ImportCache, jobs
 		maps.Copy(wl.Labels, cache.AddLabels)
 
 		if pc, found := cache.PriorityClasses[p.Spec.PriorityClassName]; found {
-			wl.Spec.PriorityClassName = pc.Name
+			wl.Spec.PriorityClassRef = kueuev1beta2.NewPodPriorityClassRef(pc.Name)
 			wl.Spec.Priority = &pc.Value
-			wl.Spec.PriorityClassSource = constants.PodPriorityClassSource
 		}
 
 		wlv1beta1 := &kueue.Workload{
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 60039d16b..5dfd7860f 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
@@ -18108,35 +18108,57 @@ spec:
                 description: |-
                   priority determines the order of access to the resources managed by the
                   ClusterQueue where the workload is queued.
-                  The priority value is populated from PriorityClassName.
+                  The priority value is populated from the referenced PriorityClass (via priorityClassRef).
                   The higher the value, the higher the priority.
-                  If priorityClassName is specified, priority must not be null.
+                  If priorityClassRef is specified, priority must not be null.
                 format: int32
                 type: integer
-              priorityClassName:
-                description: |-
-                  priorityClassName is the name of the PriorityClass the Workload is associated with.
-                  If specified, indicates the workload's priority.
-                  "system-node-critical" and "system-cluster-critical" are two special
-                  keywords which indicate the highest priorities with the former being
-                  the highest priority. Any other name must be defined by creating a
-                  PriorityClass object with that name. If not specified, the workload
-                  priority will be default or zero if there is no default.
-                maxLength: 253
-                minLength: 1
-                pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
-                type: string
-              priorityClassSource:
-                default: ""
-                description: |-
-                  priorityClassSource determines whether the priorityClass field refers to a pod PriorityClass or kueue.x-k8s.io/workloadpriorityclass.
-                  Workload's PriorityClass can accept the name of a pod priorityClass or a workloadPriorityClass.
-                  When using pod PriorityClass, a priorityClassSource field has the scheduling.k8s.io/priorityclass value.
-                enum:
-                - kueue.x-k8s.io/workloadpriorityclass
-                - scheduling.k8s.io/priorityclass
-                - ""
-                type: string
+              priorityClassRef:
+                description: priorityClassRef references a PriorityClass object that
+                  defines the workload's priority.
+                properties:
+                  group:
+                    description: |-
+                      group is the API group of the PriorityClass object.
+                      Use "kueue.x-k8s.io" for WorkloadPriorityClass.
+                      Use "scheduling.k8s.io" for Pod PriorityClass.
+                    enum:
+                    - kueue.x-k8s.io
+                    - scheduling.k8s.io
+                    type: string
+                  kind:
+                    description: kind is the kind of the PriorityClass object.
+                    enum:
+                    - WorkloadPriorityClass
+                    - PriorityClass
+                    type: string
+                  name:
+                    description: |-
+                      name is the name of the PriorityClass the Workload is associated with.
+                      If specified, indicates the workload's priority.
+                      "system-node-critical" and "system-cluster-critical" are two special
+                      keywords which indicate the highest priorities with the former being
+                      the highest priority. Any other name must be defined by creating a
+                      PriorityClass object with that name. If not specified, the workload
+                      priority will be default or zero if there is no default.
+                    maxLength: 253
+                    minLength: 1
+                    pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+                    type: string
+                required:
+                - group
+                - kind
+                - name
+                type: object
+                x-kubernetes-validations:
+                - message: only the PriorityClass kind is allowed for the scheduling.k8s.io
+                    group
+                  rule: '(self.group == ''scheduling.k8s.io'') ? self.kind == ''PriorityClass''
+                    : true'
+                - message: only the WorkloadPriorityClass kind is allowed for the
+                    kueue.x-k8s.io group
+                  rule: '(self.group == ''kueue.x-k8s.io'') ? self.kind == ''WorkloadPriorityClass''
+                    : true'
               queueName:
                 description: |-
                   queueName is the name of the LocalQueue the Workload is associated with.
@@ -18146,8 +18168,8 @@ spec:
                 type: string
             type: object
             x-kubernetes-validations:
-            - message: priority should not be nil when priorityClassName is set
-              rule: 'has(self.priorityClassName) ? has(self.priority) : true'
+            - message: priority should not be nil when priorityClassRef is set
+              rule: '!has(self.priorityClassRef) || has(self.priority)'
           status:
             description: status is the status of the Workload.
             properties:
@@ -18719,22 +18741,29 @@ spec:
             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
+        - message: priorityClassRef 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 and
-            priorityClassSource is not equal to kueue.x-k8s.io/workloadpriorityclass
+            c.type == ''QuotaReserved'' && c.status == ''True'')) ? (!has(oldSelf.spec.priorityClassRef)
+            && !has(self.spec.priorityClassRef)) || (has(oldSelf.spec.priorityClassRef)
+            && has(self.spec.priorityClassRef)) : true'
+        - message: priorityClassRef.group 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'') && (self.spec.priorityClassSource
-            != ''kueue.x-k8s.io/workloadpriorityclass'') && has(oldSelf.spec.priorityClassName)
-            && has(self.spec.priorityClassName)) ? (oldSelf.spec.priorityClassName
-            == self.spec.priorityClassName) : true'
+            c.type == ''QuotaReserved'' && c.status == ''True'') && has(oldSelf.spec.priorityClassRef)
+            && has(self.spec.priorityClassRef)) ? oldSelf.spec.priorityClassRef.group
+            == self.spec.priorityClassRef.group : true'
+        - message: priorityClassRef.kind 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.priorityClassRef)
+            && has(self.spec.priorityClassRef)) ? oldSelf.spec.priorityClassRef.kind
+            == self.spec.priorityClassRef.kind : true'
+        - message: priorityClassRef.name is immutable for scheduling.k8s.io/priorityclass
+            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.priorityClassRef)
+            && has(self.spec.priorityClassRef) && self.spec.priorityClassRef.group
+            == ''scheduling.k8s.io'' && self.spec.priorityClassRef.kind == ''PriorityClass'')
+            ? oldSelf.spec.priorityClassRef.name == self.spec.priorityClassRef.name
+            : 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)
diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go
index 856bbf038..b8b750745 100644
--- a/pkg/constants/constants.go
+++ b/pkg/constants/constants.go
@@ -32,12 +32,9 @@ const (
 	UpdatesBatchPeriod = time.Second
 
 	// DefaultPriority is used to set priority of workloads
-	// that do not specify any priority class and there is no priority class
+	// that do not specify any priority class, and there is no priority class
 	// marked as default.
-	DefaultPriority = 0
-
-	WorkloadPriorityClassSource = "kueue.x-k8s.io/workloadpriorityclass"
-	PodPriorityClassSource      = "scheduling.k8s.io/priorityclass"
+	DefaultPriority int32 = 0
 
 	DefaultPendingWorkloadsLimit = 1000
 
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index cc74f3ee1..1ef0d461f 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -52,7 +52,6 @@ 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"
@@ -388,17 +387,18 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			quotaReservedCondition := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadQuotaReserved)
 			quotaReservedWaitTime := r.clock.Since(quotaReservedCondition.LastTransitionTime.Time)
 			r.recorder.Eventf(&wl, corev1.EventTypeNormal, "Admitted", "Admitted by ClusterQueue %v, wait time since reservation was %.0fs", wl.Status.Admission.ClusterQueue, quotaReservedWaitTime.Seconds())
-			metrics.AdmittedWorkload(cqName, wl.Spec.PriorityClassName, queuedWaitTime)
-			metrics.ReportAdmissionChecksWaitTime(cqName, wl.Spec.PriorityClassName, quotaReservedWaitTime)
+			priorityClassName := workload.PriorityClassName(&wl)
+			metrics.AdmittedWorkload(cqName, priorityClassName, queuedWaitTime)
+			metrics.ReportAdmissionChecksWaitTime(cqName, priorityClassName, quotaReservedWaitTime)
 			if features.Enabled(features.LocalQueueMetrics) {
 				metrics.LocalQueueAdmittedWorkload(
 					metrics.LQRefFromWorkload(&wl),
-					wl.Spec.PriorityClassName,
+					priorityClassName,
 					queuedWaitTime,
 				)
 				metrics.ReportLocalQueueAdmissionChecksWaitTime(
 					metrics.LQRefFromWorkload(&wl),
-					wl.Spec.PriorityClassName,
+					priorityClassName,
 					quotaReservedWaitTime,
 				)
 			}
@@ -924,10 +924,9 @@ func (r *WorkloadReconciler) Update(e event.TypedUpdateEvent[*kueue.Workload]) b
 }
 
 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
+	return workload.IsWorkloadPriorityClass(old) && workload.IsWorkloadPriorityClass(new) &&
+		workload.PriorityClassName(old) != "" && workload.PriorityClassName(new) != "" &&
+		workload.PriorityClassName(old) != workload.PriorityClassName(new)
 }
 
 func (r *WorkloadReconciler) Generic(e event.TypedGenericEvent[*kueue.Workload]) bool {
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index c1830f7bc..3a3c8df76 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -527,14 +527,15 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 			// update the metrics only when PodsReady condition status is true
 			if condition.Status == metav1.ConditionTrue {
 				cqName := wl.Status.Admission.ClusterQueue
+				priorityClassName := workload.PriorityClassName(wl)
 				queuedUntilReadyWaitTime := workload.QueuedWaitTime(wl, r.clock)
-				metrics.ReadyWaitTime(cqName, wl.Spec.PriorityClassName, queuedUntilReadyWaitTime)
+				metrics.ReadyWaitTime(cqName, priorityClassName, queuedUntilReadyWaitTime)
 				admittedCond := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadAdmitted)
 				admittedUntilReadyWaitTime := condition.LastTransitionTime.Sub(admittedCond.LastTransitionTime.Time)
-				metrics.ReportAdmittedUntilReadyWaitTime(cqName, wl.Spec.PriorityClassName, admittedUntilReadyWaitTime)
+				metrics.ReportAdmittedUntilReadyWaitTime(cqName, priorityClassName, admittedUntilReadyWaitTime)
 				if features.Enabled(features.LocalQueueMetrics) {
-					metrics.LocalQueueReadyWaitTime(metrics.LQRefFromWorkload(wl), wl.Spec.PriorityClassName, queuedUntilReadyWaitTime)
-					metrics.ReportLocalQueueAdmittedUntilReadyWaitTime(metrics.LQRefFromWorkload(wl), wl.Spec.PriorityClassName, admittedUntilReadyWaitTime)
+					metrics.LocalQueueReadyWaitTime(metrics.LQRefFromWorkload(wl), priorityClassName, queuedUntilReadyWaitTime)
+					metrics.ReportLocalQueueAdmittedUntilReadyWaitTime(metrics.LQRefFromWorkload(wl), priorityClassName, admittedUntilReadyWaitTime)
 				}
 			}
 			return ctrl.Result{}, nil
@@ -945,7 +946,7 @@ func (r *JobReconciler) ensureOneWorkload(ctx context.Context, job GenericJob, o
 
 // 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 workload.IsWorkloadPriorityClass(wl) && WorkloadPriorityClassName(obj) != workload.PriorityClassName(wl) {
 		if err := PrepareWorkloadPriority(ctx, c, obj, wl, customPriorityClassFunc); err != nil {
 			return fmt.Errorf("prepare workload priority: %w", err)
 		}
@@ -1294,14 +1295,13 @@ func getCustomPriorityClassFuncFromJob(job GenericJob) func() string {
 }
 
 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)
+	priorityClassRef, priority, err := ExtractPriority(ctx, c, obj, wl.Spec.PodSets, customPriorityClassFunc)
 	if err != nil {
 		return err
 	}
 
-	wl.Spec.PriorityClassName = priorityClassName
-	wl.Spec.Priority = &p
-	wl.Spec.PriorityClassSource = source
+	wl.Spec.PriorityClassRef = priorityClassRef
+	wl.Spec.Priority = &priority
 
 	return nil
 }
@@ -1320,7 +1320,7 @@ func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl
 	return nil
 }
 
-func ExtractPriority(ctx context.Context, c client.Client, obj client.Object, podSets []kueue.PodSet, customPriorityClassFunc func() string) (string, string, int32, error) {
+func ExtractPriority(ctx context.Context, c client.Client, obj client.Object, podSets []kueue.PodSet, customPriorityClassFunc func() string) (*kueue.PriorityClassRef, int32, error) {
 	if workloadPriorityClass := WorkloadPriorityClassName(obj); len(workloadPriorityClass) > 0 {
 		return utilpriority.GetPriorityFromWorkloadPriorityClass(ctx, c, workloadPriorityClass)
 	}
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index f2ab75e6a..d7c8cae84 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -807,9 +807,10 @@ func (s *Scheduler) recordQuotaReservationMetrics(newWorkload, originalWorkload
 
 	s.recorder.Eventf(newWorkload, corev1.EventTypeNormal, "QuotaReserved", "Quota reserved in ClusterQueue %v, wait time since queued was %.0fs", admission.ClusterQueue, waitTime.Seconds())
 
-	metrics.QuotaReservedWorkload(admission.ClusterQueue, newWorkload.Spec.PriorityClassName, waitTime)
+	priorityClassName := workload.PriorityClassName(newWorkload)
+	metrics.QuotaReservedWorkload(admission.ClusterQueue, priorityClassName, waitTime)
 	if features.Enabled(features.LocalQueueMetrics) {
-		metrics.LocalQueueQuotaReservedWorkload(metrics.LQRefFromWorkload(newWorkload), newWorkload.Spec.PriorityClassName, waitTime)
+		metrics.LocalQueueQuotaReservedWorkload(metrics.LQRefFromWorkload(newWorkload), priorityClassName, waitTime)
 	}
 }
 
@@ -820,20 +821,17 @@ func (s *Scheduler) recordWorkloadAdmissionEvents(newWorkload, originalWorkload
 	}
 
 	s.recorder.Eventf(newWorkload, corev1.EventTypeNormal, "Admitted", "Admitted by ClusterQueue %v, wait time since reservation was 0s", admission.ClusterQueue)
-	metrics.AdmittedWorkload(admission.ClusterQueue, newWorkload.Spec.PriorityClassName, waitTime)
 
+	priorityClassName := workload.PriorityClassName(newWorkload)
+	metrics.AdmittedWorkload(admission.ClusterQueue, priorityClassName, waitTime)
 	if features.Enabled(features.LocalQueueMetrics) {
-		metrics.LocalQueueAdmittedWorkload(
-			metrics.LQRefFromWorkload(newWorkload),
-			newWorkload.Spec.PriorityClassName,
-			waitTime,
-		)
+		metrics.LocalQueueAdmittedWorkload(metrics.LQRefFromWorkload(newWorkload), priorityClassName, waitTime)
 	}
 
 	if len(newWorkload.Status.AdmissionChecks) > 0 {
-		metrics.ReportAdmissionChecksWaitTime(admission.ClusterQueue, newWorkload.Spec.PriorityClassName, 0)
+		metrics.ReportAdmissionChecksWaitTime(admission.ClusterQueue, priorityClassName, 0)
 		if features.Enabled(features.LocalQueueMetrics) {
-			metrics.ReportLocalQueueAdmissionChecksWaitTime(metrics.LQRefFromWorkload(newWorkload), newWorkload.Spec.PriorityClassName, 0)
+			metrics.ReportLocalQueueAdmissionChecksWaitTime(metrics.LQRefFromWorkload(newWorkload), priorityClassName, 0)
 		}
 	}
 }
diff --git a/pkg/util/priority/priority.go b/pkg/util/priority/priority.go
index 060d896d8..ee4894c3b 100644
--- a/pkg/util/priority/priority.go
+++ b/pkg/util/priority/priority.go
@@ -30,51 +30,49 @@ import (
 
 // Priority returns priority of the given workload.
 func Priority(w *kueue.Workload) int32 {
-	// When priority of a running workload is nil, it means it was created at a time
+	// When the priority of a running workload is nil, it means it was created at a time
 	// that there was no global default priority class and the priority class
 	// name of the pod was empty. So, we resolve to the static default priority.
 	return ptr.Deref(w.Spec.Priority, constants.DefaultPriority)
 }
 
 // GetPriorityFromPriorityClass returns the priority populated from
-// priority class. If not specified, priority will be default or
+// priority class. If not specified, the priority will be default or
 // zero if there is no default.
-func GetPriorityFromPriorityClass(ctx context.Context, client client.Client,
-	priorityClass string) (string, string, int32, error) {
+func GetPriorityFromPriorityClass(ctx context.Context, client client.Client, priorityClass string) (*kueue.PriorityClassRef, int32, error) {
 	if len(priorityClass) == 0 {
 		return getDefaultPriority(ctx, client)
 	}
 
 	pc := &schedulingv1.PriorityClass{}
 	if err := client.Get(ctx, types.NamespacedName{Name: priorityClass}, pc); err != nil {
-		return "", "", 0, err
+		return nil, 0, err
 	}
 
-	return pc.Name, constants.PodPriorityClassSource, pc.Value, nil
+	return kueue.NewPodPriorityClassRef(pc.Name), pc.Value, nil
 }
 
 // GetPriorityFromWorkloadPriorityClass returns the priority populated from
 // workload priority class. If not specified, returns 0.
 // DefaultPriority is not called within this function
-// because k8s priority class should be  checked next.
-func GetPriorityFromWorkloadPriorityClass(ctx context.Context, client client.Client,
-	workloadPriorityClass string) (string, string, int32, error) {
+// because k8s priority class should be checked next.
+func GetPriorityFromWorkloadPriorityClass(ctx context.Context, client client.Client, workloadPriorityClass string) (*kueue.PriorityClassRef, int32, error) {
 	wpc := &kueue.WorkloadPriorityClass{}
 	if err := client.Get(ctx, types.NamespacedName{Name: workloadPriorityClass}, wpc); err != nil {
-		return "", "", 0, err
+		return nil, 0, err
 	}
-	return wpc.Name, constants.WorkloadPriorityClassSource, wpc.Value, nil
+	return kueue.NewWorkloadPriorityClassRef(wpc.Name), wpc.Value, nil
 }
 
-func getDefaultPriority(ctx context.Context, client client.Client) (string, string, int32, error) {
+func getDefaultPriority(ctx context.Context, client client.Client) (*kueue.PriorityClassRef, int32, error) {
 	dpc, err := getDefaultPriorityClass(ctx, client)
 	if err != nil {
-		return "", "", 0, err
+		return nil, 0, err
 	}
 	if dpc != nil {
-		return dpc.Name, constants.PodPriorityClassSource, dpc.Value, nil
+		return kueue.NewPodPriorityClassRef(dpc.Name), dpc.Value, nil
 	}
-	return "", "", int32(constants.DefaultPriority), nil
+	return nil, constants.DefaultPriority, nil
 }
 
 func getDefaultPriorityClass(ctx context.Context, client client.Client) (*schedulingv1.PriorityClass, error) {
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 5ef39aa89..d465a11ac 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -1296,11 +1296,24 @@ func Evict(ctx context.Context, c client.Client, recorder record.EventRecorder,
 	}
 	reportEvictedWorkload(recorder, wl, wl.Status.Admission.ClusterQueue, reason, msg, underlyingCause)
 	if reportWorkloadEvictedOnce {
-		metrics.ReportEvictedWorkloadsOnce(wl.Status.Admission.ClusterQueue, reason, string(underlyingCause), wl.Spec.PriorityClassName)
+		metrics.ReportEvictedWorkloadsOnce(wl.Status.Admission.ClusterQueue, reason, string(underlyingCause), PriorityClassName(wl))
 	}
 	return nil
 }
 
+func PriorityClassName(wl *kueue.Workload) string {
+	if wl.Spec.PriorityClassRef != nil {
+		return wl.Spec.PriorityClassRef.Name
+	}
+	return ""
+}
+
+func IsWorkloadPriorityClass(wl *kueue.Workload) bool {
+	return wl.Spec.PriorityClassRef != nil &&
+		wl.Spec.PriorityClassRef.Kind == kueue.WorkloadPriorityClassKind &&
+		wl.Spec.PriorityClassRef.Group == kueue.WorkloadPriorityClassGroup
+}
+
 func prepareForEviction(w *kueue.Workload, now time.Time, reason, message string) {
 	SetEvictedCondition(w, now, reason, message)
 	resetClusterNomination(w)
@@ -1318,7 +1331,8 @@ func resetUnhealthyNodes(w *kueue.Workload) {
 }
 
 func reportEvictedWorkload(recorder record.EventRecorder, wl *kueue.Workload, cqName kueue.ClusterQueueReference, reason, message string, underlyingCause kueue.EvictionUnderlyingCause) {
-	metrics.ReportEvictedWorkloads(cqName, reason, string(underlyingCause), wl.Spec.PriorityClassName)
+	priorityClassName := PriorityClassName(wl)
+	metrics.ReportEvictedWorkloads(cqName, reason, string(underlyingCause), priorityClassName)
 	if podsReadyToEvictionTime := workloadsWithPodsReadyToEvictedTime(wl); podsReadyToEvictionTime != nil {
 		metrics.PodsReadyToEvictedTimeSeconds.WithLabelValues(string(cqName), reason, string(underlyingCause)).Observe(podsReadyToEvictionTime.Seconds())
 	}
@@ -1327,7 +1341,7 @@ func reportEvictedWorkload(recorder record.EventRecorder, wl *kueue.Workload, cq
 			metrics.LQRefFromWorkload(wl),
 			reason,
 			string(underlyingCause),
-			wl.Spec.PriorityClassName,
+			priorityClassName,
 		)
 	}
 	eventReason := ReasonWithCause(kueue.WorkloadEvicted, reason)
diff --git a/site/content/en/docs/reference/kueue.v1beta2.md b/site/content/en/docs/reference/kueue.v1beta2.md
index dc6b1ef17..4a061fcd5 100644
--- a/site/content/en/docs/reference/kueue.v1beta2.md
+++ b/site/content/en/docs/reference/kueue.v1beta2.md
@@ -2200,6 +2200,82 @@ result in failure during workload admission.</p>
 
 
 
+## `PriorityClassGroup`     {#kueue-x-k8s-io-v1beta2-PriorityClassGroup}
+    
+(Alias of `string`)
+
+**Appears in:**
+
+- [PriorityClassRef](#kueue-x-k8s-io-v1beta2-PriorityClassRef)
+
+
+<p>PriorityClassGroup indicates the API group of the PriorityClass object.</p>
+
+
+
+
+## `PriorityClassKind`     {#kueue-x-k8s-io-v1beta2-PriorityClassKind}
+    
+(Alias of `string`)
+
+**Appears in:**
+
+- [PriorityClassRef](#kueue-x-k8s-io-v1beta2-PriorityClassRef)
+
+
+<p>PriorityClassKind is the kind of the PriorityClass object.</p>
+
+
+
+
+## `PriorityClassRef`     {#kueue-x-k8s-io-v1beta2-PriorityClassRef}
+    
+
+**Appears in:**
+
+- [WorkloadSpec](#kueue-x-k8s-io-v1beta2-WorkloadSpec)
+
+
+<p>PriorityClassRef references a PriorityClass in a specific API group.</p>
+
+
+<table class="table">
+<thead><tr><th width="30%">Field</th><th>Description</th></tr></thead>
+<tbody>
+    
+  
+<tr><td><code>group</code> <B>[Required]</B><br/>
+<a href="#kueue-x-k8s-io-v1beta2-PriorityClassGroup"><code>PriorityClassGroup</code></a>
+</td>
+<td>
+   <p>group is the API group of the PriorityClass object.
+Use &quot;kueue.x-k8s.io&quot; for WorkloadPriorityClass.
+Use &quot;scheduling.k8s.io&quot; for Pod PriorityClass.</p>
+</td>
+</tr>
+<tr><td><code>kind</code> <B>[Required]</B><br/>
+<a href="#kueue-x-k8s-io-v1beta2-PriorityClassKind"><code>PriorityClassKind</code></a>
+</td>
+<td>
+   <p>kind is the kind of the PriorityClass object.</p>
+</td>
+</tr>
+<tr><td><code>name</code> <B>[Required]</B><br/>
+<code>string</code>
+</td>
+<td>
+   <p>name is the name of the PriorityClass the Workload is associated with.
+If specified, indicates the workload's priority.
+&quot;system-node-critical&quot; and &quot;system-cluster-critical&quot; are two special
+keywords which indicate the highest priorities with the former being
+the highest priority. Any other name must be defined by creating a
+PriorityClass object with that name. If not specified, the workload
+priority will be default or zero if there is no default.</p>
+</td>
+</tr>
+</tbody>
+</table>
+
 ## `ProvisioningRequestConfigPodSetMergePolicy`     {#kueue-x-k8s-io-v1beta2-ProvisioningRequestConfigPodSetMergePolicy}
     
 (Alias of `string`)
@@ -2998,17 +3074,11 @@ podSets cannot be changed.</p>
 queueName cannot be changed while .status.admission is not null.</p>
 </td>
 </tr>
-<tr><td><code>priorityClassName</code><br/>
-<code>string</code>
+<tr><td><code>priorityClassRef</code><br/>
+<a href="#kueue-x-k8s-io-v1beta2-PriorityClassRef"><code>PriorityClassRef</code></a>
 </td>
 <td>
-   <p>priorityClassName is the name of the PriorityClass the Workload is associated with.
-If specified, indicates the workload's priority.
-&quot;system-node-critical&quot; and &quot;system-cluster-critical&quot; are two special
-keywords which indicate the highest priorities with the former being
-the highest priority. Any other name must be defined by creating a
-PriorityClass object with that name. If not specified, the workload
-priority will be default or zero if there is no default.</p>
+   <p>priorityClassRef references a PriorityClass object that defines the workload's priority.</p>
 </td>
 </tr>
 <tr><td><code>priority</code><br/>
@@ -3017,18 +3087,9 @@ priority will be default or zero if there is no default.</p>
 <td>
    <p>priority determines the order of access to the resources managed by the
 ClusterQueue where the workload is queued.
-The priority value is populated from PriorityClassName.
+The priority value is populated from the referenced PriorityClass (via priorityClassRef).
 The higher the value, the higher the priority.
-If priorityClassName is specified, priority must not be null.</p>
-</td>
-</tr>
-<tr><td><code>priorityClassSource</code><br/>
-<code>string</code>
-</td>
-<td>
-   <p>priorityClassSource determines whether the priorityClass field refers to a pod PriorityClass or kueue.x-k8s.io/workloadpriorityclass.
-Workload's PriorityClass can accept the name of a pod priorityClass or a workloadPriorityClass.
-When using pod PriorityClass, a priorityClassSource field has the scheduling.k8s.io/priorityclass value.</p>
+If priorityClassRef is specified, priority must not be null.</p>
 </td>
 </tr>
 <tr><td><code>active</code> <B>[Required]</B><br/>



Test patch:
diff --git a/apis/kueue/v1beta1/workload_conversion_test.go b/apis/kueue/v1beta1/workload_conversion_test.go
index 6c90084e9..e9b9bdd0a 100644
--- a/apis/kueue/v1beta1/workload_conversion_test.go
+++ b/apis/kueue/v1beta1/workload_conversion_test.go
@@ -106,6 +106,48 @@ func TestWorkloadConvertTo(t *testing.T) {
 				},
 			},
 		},
+		"with PodPriorityClassSource": {
+			input: &Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: WorkloadSpec{
+					Priority:            ptr.To[int32](100),
+					PriorityClassSource: PodPriorityClassSource,
+					PriorityClassName:   "low",
+				},
+			},
+			expected: &v1beta2.Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: v1beta2.WorkloadSpec{
+					Priority: ptr.To[int32](100),
+					PriorityClassRef: &v1beta2.PriorityClassRef{
+						Group: v1beta2.PodPriorityClassGroup,
+						Kind:  v1beta2.PodPriorityClassKind,
+						Name:  "low",
+					},
+				},
+			},
+		},
+		"with WorkloadPriorityClassSource": {
+			input: &Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: WorkloadSpec{
+					Priority:            ptr.To[int32](100),
+					PriorityClassSource: WorkloadPriorityClassSource,
+					PriorityClassName:   "low",
+				},
+			},
+			expected: &v1beta2.Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: v1beta2.WorkloadSpec{
+					Priority: ptr.To[int32](100),
+					PriorityClassRef: &v1beta2.PriorityClassRef{
+						Group: v1beta2.WorkloadPriorityClassGroup,
+						Kind:  v1beta2.WorkloadPriorityClassKind,
+						Name:  "low",
+					},
+				},
+			},
+		},
 	}
 
 	for name, tc := range testCases {
@@ -201,6 +243,48 @@ func TestWorkloadConvertFrom(t *testing.T) {
 				},
 			},
 		},
+		"with PodPriorityClassRef": {
+			input: &v1beta2.Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: v1beta2.WorkloadSpec{
+					Priority: ptr.To[int32](100),
+					PriorityClassRef: &v1beta2.PriorityClassRef{
+						Group: v1beta2.PodPriorityClassGroup,
+						Kind:  v1beta2.PodPriorityClassKind,
+						Name:  "low",
+					},
+				},
+			},
+			expected: &Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: WorkloadSpec{
+					Priority:            ptr.To[int32](100),
+					PriorityClassSource: PodPriorityClassSource,
+					PriorityClassName:   "low",
+				},
+			},
+		},
+		"with WorkloadPriorityClassRef": {
+			input: &v1beta2.Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: v1beta2.WorkloadSpec{
+					Priority: ptr.To[int32](100),
+					PriorityClassRef: &v1beta2.PriorityClassRef{
+						Group: v1beta2.WorkloadPriorityClassGroup,
+						Kind:  v1beta2.WorkloadPriorityClassKind,
+						Name:  "low",
+					},
+				},
+			},
+			expected: &Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: WorkloadSpec{
+					Priority:            ptr.To[int32](100),
+					PriorityClassSource: WorkloadPriorityClassSource,
+					PriorityClassName:   "low",
+				},
+			},
+		},
 	}
 
 	for name, tc := range testCases {
@@ -217,15 +301,17 @@ func TestWorkloadConvertFrom(t *testing.T) {
 }
 
 func TestWorkloadConversion_RoundTrip(t *testing.T) {
+	defaultObjectMeta := metav1.ObjectMeta{
+		Name:      "test-workload",
+		Namespace: "default",
+	}
+
 	testCases := map[string]struct {
 		v1beta1Obj *Workload
 	}{
 		"complete Workload with AccumulatedPastExexcutionTimeSeconds": {
 			v1beta1Obj: &Workload{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "test-workload",
-					Namespace: "default",
-				},
+				ObjectMeta: defaultObjectMeta,
 				Status: WorkloadStatus{
 					Conditions: []metav1.Condition{
 						{
@@ -240,10 +326,7 @@ func TestWorkloadConversion_RoundTrip(t *testing.T) {
 		},
 		"Workload with nil AccumulatedPastExexcutionTimeSeconds": {
 			v1beta1Obj: &Workload{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "simple-workload",
-					Namespace: "test-ns",
-				},
+				ObjectMeta: defaultObjectMeta,
 				Status: WorkloadStatus{
 					AccumulatedPastExexcutionTimeSeconds: nil,
 				},
@@ -251,15 +334,32 @@ func TestWorkloadConversion_RoundTrip(t *testing.T) {
 		},
 		"Workload with zero AccumulatedPastExexcutionTimeSeconds": {
 			v1beta1Obj: &Workload{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "zero-workload",
-					Namespace: "test-ns",
-				},
+				ObjectMeta: defaultObjectMeta,
 				Status: WorkloadStatus{
 					AccumulatedPastExexcutionTimeSeconds: ptr.To[int32](0),
 				},
 			},
 		},
+		"Workload with PodPriorityClassSource": {
+			v1beta1Obj: &Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: WorkloadSpec{
+					Priority:            ptr.To[int32](100),
+					PriorityClassSource: PodPriorityClassSource,
+					PriorityClassName:   "low",
+				},
+			},
+		},
+		"Workload with WorkloadPriorityClassSource": {
+			v1beta1Obj: &Workload{
+				ObjectMeta: defaultObjectMeta,
+				Spec: WorkloadSpec{
+					Priority:            ptr.To[int32](100),
+					PriorityClassSource: WorkloadPriorityClassSource,
+					PriorityClassName:   "low",
+				},
+			},
+		},
 	}
 
 	for name, tc := range testCases {
diff --git a/pkg/cache/queue/cluster_queue_test.go b/pkg/cache/queue/cluster_queue_test.go
index 5544fc3a9..2d79a05f6 100644
--- a/pkg/cache/queue/cluster_queue_test.go
+++ b/pkg/cache/queue/cluster_queue_test.go
@@ -728,12 +728,12 @@ func TestStrictFIFO(t *testing.T) {
 			name: "w1.priority is higher than w2.priority",
 			w1: utiltestingapi.MakeWorkload("w1", "").
 				Creation(t1).
-				PriorityClass("highPriority").
+				PodPriorityClassRef("highPriority").
 				Priority(highPriority).
 				Obj(),
 			w2: utiltestingapi.MakeWorkload("w2", "").
 				Creation(t2).
-				PriorityClass("lowPriority").
+				PodPriorityClassRef("lowPriority").
 				Priority(lowPriority).
 				Obj(),
 			expected: "w1",
@@ -789,12 +789,12 @@ func TestStrictFIFO(t *testing.T) {
 			name: "p1.priority is lower than p2.priority and w1.create time is earlier than w2.create time",
 			w1: utiltestingapi.MakeWorkload("w1", "").
 				Creation(t1).
-				PriorityClass("lowPriority").
+				PodPriorityClassRef("lowPriority").
 				Priority(lowPriority).
 				Obj(),
 			w2: utiltestingapi.MakeWorkload("w2", "").
 				Creation(t2).
-				PriorityClass("highPriority").
+				PodPriorityClassRef("highPriority").
 				Priority(highPriority).
 				Obj(),
 			expected: "w2",
diff --git a/pkg/controller/jobs/job/job_controller_test.go b/pkg/controller/jobs/job/job_controller_test.go
index fc02ee2ba..89115e213 100644
--- a/pkg/controller/jobs/job/job_controller_test.go
+++ b/pkg/controller/jobs/job/job_controller_test.go
@@ -38,7 +38,6 @@ import (
 
 	configapi "sigs.k8s.io/kueue/apis/config/v1beta2"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
-	"sigs.k8s.io/kueue/pkg/constants"
 	controllerconsts "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -2689,8 +2688,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").Obj()).
 					Queue("foo").
 					Priority(baseWPCWrapper.Value).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
-					PriorityClass(baseWPCWrapper.Name).
+					WorkloadPriorityClassRef(baseWPCWrapper.Name).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -2702,8 +2700,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").Obj()).
 					Queue("foo").
 					Priority(highWPCWrapper.Value).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
-					PriorityClass(highWPCWrapper.Name).
+					WorkloadPriorityClassRef(highWPCWrapper.Name).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -2739,8 +2736,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).PriorityClass(basePCWrapper.Name).Request(corev1.ResourceCPU, "1").Obj()).
 					Queue("foo").
 					Priority(basePCWrapper.Value).
-					PriorityClassSource(constants.PodPriorityClassSource).
-					PriorityClass(basePCWrapper.Name).
+					PodPriorityClassRef(basePCWrapper.Name).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -2752,8 +2748,7 @@ func TestReconciler(t *testing.T) {
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).PriorityClass(basePCWrapper.Name).Request(corev1.ResourceCPU, "1").Obj()).
 					Queue("foo").
 					Priority(basePCWrapper.Value).
-					PriorityClassSource(constants.PodPriorityClassSource).
-					PriorityClass(basePCWrapper.Name).
+					PodPriorityClassRef(basePCWrapper.Name).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -3185,9 +3180,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -3230,9 +3224,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-pc").
+					PodPriorityClassRef("test-pc").
 					Priority(200).
-					PriorityClassSource(constants.PodPriorityClassSource).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -3277,9 +3270,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -3349,9 +3341,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Obj(),
 			},
 			wantWorkloads: []kueue.Workload{
@@ -3359,9 +3350,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
@@ -3394,9 +3384,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "other-job", "other-uid").
 					Obj(),
 			},
@@ -3405,9 +3394,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 10).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "other-job", "other-uid").
 					Obj(),
 			},
@@ -3438,9 +3426,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Obj(),
 			},
 			wantWorkloads: []kueue.Workload{
@@ -3448,9 +3435,8 @@ func TestReconciler(t *testing.T) {
 					Finalizers(kueue.ResourceInUseFinalizerName).
 					PodSets(*utiltestingapi.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").PriorityClass("test-pc").Obj()).
 					Queue("test-queue").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Labels(map[string]string{
 						controllerconsts.JobUIDLabel: "test-uid",
 					}).
diff --git a/pkg/controller/jobs/jobset/jobset_controller_test.go b/pkg/controller/jobs/jobset/jobset_controller_test.go
index b033317de..96dee2e6e 100644
--- a/pkg/controller/jobs/jobset/jobset_controller_test.go
+++ b/pkg/controller/jobs/jobset/jobset_controller_test.go
@@ -32,7 +32,6 @@ import (
 	jobset "sigs.k8s.io/jobset/api/jobset/v1alpha2"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
-	"sigs.k8s.io/kueue/pkg/constants"
 	controllerconsts "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -473,9 +472,8 @@ func TestReconciler(t *testing.T) {
 			).WorkloadPriorityClass("test-wpc").Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltestingapi.MakeWorkload("jobset", "ns").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					PodSets(
 						*utiltestingapi.MakePodSet("replicated-job-1", 1).
 							PodIndexLabel(ptr.To("batch.kubernetes.io/job-completion-index")).
@@ -512,9 +510,8 @@ func TestReconciler(t *testing.T) {
 			).PriorityClass("test-pc").Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltestingapi.MakeWorkload("jobset", "ns").
-					PriorityClass("test-pc").
+					PodPriorityClassRef("test-pc").
 					Priority(200).
-					PriorityClassSource(constants.PodPriorityClassSource).
 					PodSets(
 						*utiltestingapi.MakePodSet("replicated-job-1", 1).
 							PodIndexLabel(ptr.To("batch.kubernetes.io/job-completion-index")).
@@ -551,9 +548,8 @@ func TestReconciler(t *testing.T) {
 			).PriorityClass("test-pc").WorkloadPriorityClass("test-wpc").Obj(),
 			wantWorkloads: []kueue.Workload{
 				*utiltestingapi.MakeWorkload("jobset", "ns").
-					PriorityClass("test-wpc").
+					WorkloadPriorityClassRef("test-wpc").
 					Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					PodSets(
 						*utiltestingapi.MakePodSet("replicated-job-1", 1).
 							PodIndexLabel(ptr.To("batch.kubernetes.io/job-completion-index")).
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
index 2fbe671eb..63e5ed9df 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler_test.go
@@ -32,7 +32,6 @@ import (
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
-	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -533,9 +532,8 @@ func TestReconciler(t *testing.T) {
 							Count:           1,
 							TopologyRequest: nil,
 						}).
-					PriorityClass("high-priority").
+					WorkloadPriorityClassRef("high-priority").
 					Priority(5000).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
 					Obj(),
 			},
 			wantEvents: []utiltesting.EventRecord{
diff --git a/pkg/controller/jobs/mpijob/mpijob_controller_test.go b/pkg/controller/jobs/mpijob/mpijob_controller_test.go
index 2fb69af05..1c752ada5 100644
--- a/pkg/controller/jobs/mpijob/mpijob_controller_test.go
+++ b/pkg/controller/jobs/mpijob/mpijob_controller_test.go
@@ -31,7 +31,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
-	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/features"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -408,8 +407,9 @@ func TestReconciler(t *testing.T) {
 						*utiltestingapi.MakePodSet("worker", 2).
 							PodIndexLabel(ptr.To(kfmpi.ReplicaIndexLabel)).
 							Obj(),
-					).PriorityClass("test-wpc").Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
+					).
+					WorkloadPriorityClassRef("test-wpc").
+					Priority(100).
 					Obj(),
 			},
 		},
@@ -432,8 +432,9 @@ func TestReconciler(t *testing.T) {
 						*utiltestingapi.MakePodSet("worker", 2).
 							PodIndexLabel(ptr.To(kfmpi.ReplicaIndexLabel)).
 							Obj(),
-					).PriorityClass("test-pc").Priority(200).
-					PriorityClassSource(constants.PodPriorityClassSource).
+					).
+					PodPriorityClassRef("test-pc").
+					Priority(200).
 					Obj(),
 			},
 		},
@@ -458,8 +459,9 @@ func TestReconciler(t *testing.T) {
 						*utiltestingapi.MakePodSet("worker", 2).
 							PodIndexLabel(ptr.To(kfmpi.ReplicaIndexLabel)).
 							Obj(),
-					).PriorityClass("test-wpc").Priority(100).
-					PriorityClassSource(constants.WorkloadPriorityClassSource).
+					).
+					WorkloadPriorityClassRef("test-wpc").
+					Priority(100).
 					Obj(),
 			},
 		},
diff --git a/pkg/util/priority/priority_test.go b/pkg/util/priority/priority_test.go
index 6d8f005d5..26ca76df6 100644
--- a/pkg/util/priority/priority_test.go
+++ b/pkg/util/priority/priority_test.go
@@ -66,12 +66,11 @@ func TestGetPriorityFromPriorityClass(t *testing.T) {
 	}
 
 	tests := map[string]struct {
-		priorityClassList       *schedulingv1.PriorityClassList
-		priorityClassName       string
-		wantPriorityClassName   string
-		wantPriorityClassSource string
-		wantPriorityClassValue  int32
-		wantErr                 error
+		priorityClassList      *schedulingv1.PriorityClassList
+		priorityClassName      string
+		wantPriorityClassRef   *kueue.PriorityClassRef
+		wantPriorityClassValue int32
+		wantErr                error
 	}{
 		"priorityClass is specified and it exists": {
 			priorityClassList: &schedulingv1.PriorityClassList{
@@ -82,10 +81,10 @@ func TestGetPriorityFromPriorityClass(t *testing.T) {
 					},
 				},
 			},
-			priorityClassName:       "test",
-			wantPriorityClassSource: constants.PodPriorityClassSource,
-			wantPriorityClassName:   "test",
-			wantPriorityClassValue:  50,
+
+			priorityClassName:      "test",
+			wantPriorityClassRef:   kueue.NewPodPriorityClassRef("test"),
+			wantPriorityClassValue: 50,
 		},
 		"priorityClass is specified and it does not exist": {
 			priorityClassList: &schedulingv1.PriorityClassList{
@@ -104,9 +103,8 @@ func TestGetPriorityFromPriorityClass(t *testing.T) {
 					},
 				},
 			},
-			wantPriorityClassName:   "globalDefault",
-			wantPriorityClassSource: constants.PodPriorityClassSource,
-			wantPriorityClassValue:  40,
+			wantPriorityClassRef:   kueue.NewPodPriorityClassRef("globalDefault"),
+			wantPriorityClassValue: 40,
 		},
 		"priorityClass is unspecified and multiple global defaults exist": {
 			priorityClassList: &schedulingv1.PriorityClassList{
@@ -128,9 +126,8 @@ func TestGetPriorityFromPriorityClass(t *testing.T) {
 					},
 				},
 			},
-			wantPriorityClassName:   "globalDefault2",
-			wantPriorityClassSource: constants.PodPriorityClassSource,
-			wantPriorityClassValue:  20,
+			wantPriorityClassRef:   kueue.NewPodPriorityClassRef("globalDefault2"),
+			wantPriorityClassValue: 20,
 		},
 	}
 
@@ -142,17 +139,13 @@ func TestGetPriorityFromPriorityClass(t *testing.T) {
 			client := builder.Build()
 
 			ctx, _ := utiltesting.ContextWithLog(t)
-			name, source, value, err := GetPriorityFromPriorityClass(ctx, client, tt.priorityClassName)
+			priorityClassRef, value, err := GetPriorityFromPriorityClass(ctx, client, tt.priorityClassName)
 			if diff := cmp.Diff(tt.wantErr, err); diff != "" {
 				t.Errorf("unexpected error (-want,+got):\n%s", diff)
 			}
 
-			if name != tt.wantPriorityClassName {
-				t.Errorf("unexpected name: got: %s, expected: %s", name, tt.wantPriorityClassName)
-			}
-
-			if source != tt.wantPriorityClassSource {
-				t.Errorf("unexpected source: got: %s, expected: %s", source, tt.wantPriorityClassSource)
+			if diff := cmp.Diff(tt.wantPriorityClassRef, priorityClassRef); diff != "" {
+				t.Errorf("unexpected priortyClassRef (-want,+got):\n%s", diff)
 			}
 
 			if value != tt.wantPriorityClassValue {
@@ -169,12 +162,11 @@ func TestGetPriorityFromWorkloadPriorityClass(t *testing.T) {
 	}
 
 	tests := map[string]struct {
-		workloadPriorityClassList       *kueue.WorkloadPriorityClassList
-		workloadPriorityClassName       string
-		wantWorkloadPriorityClassName   string
-		wantWorkloadPriorityClassSource string
-		wantWorkloadPriorityClassValue  int32
-		wantErr                         error
+		workloadPriorityClassList *kueue.WorkloadPriorityClassList
+		workloadPriorityClassName string
+		wantPriorityClassRef      *kueue.PriorityClassRef
+		wantPriorityClassValue    int32
+		wantErr                   error
 	}{
 		"workloadPriorityClass is specified and it exists": {
 			workloadPriorityClassList: &kueue.WorkloadPriorityClassList{
@@ -185,10 +177,9 @@ func TestGetPriorityFromWorkloadPriorityClass(t *testing.T) {
 					},
 				},
 			},
-			workloadPriorityClassName:       "test",
-			wantWorkloadPriorityClassSource: constants.WorkloadPriorityClassSource,
-			wantWorkloadPriorityClassName:   "test",
-			wantWorkloadPriorityClassValue:  50,
+			workloadPriorityClassName: "test",
+			wantPriorityClassRef:      kueue.NewWorkloadPriorityClassRef("test"),
+			wantPriorityClassValue:    50,
 		},
 		"workloadPriorityClass is specified and it does not exist": {
 			workloadPriorityClassList: &kueue.WorkloadPriorityClassList{
@@ -206,21 +197,17 @@ func TestGetPriorityFromWorkloadPriorityClass(t *testing.T) {
 			builder := fake.NewClientBuilder().WithScheme(scheme).WithLists(tt.workloadPriorityClassList)
 			client := builder.Build()
 			ctx, _ := utiltesting.ContextWithLog(t)
-			name, source, value, err := GetPriorityFromWorkloadPriorityClass(ctx, client, tt.workloadPriorityClassName)
+			priorityClassRef, value, err := GetPriorityFromWorkloadPriorityClass(ctx, client, tt.workloadPriorityClassName)
 			if diff := cmp.Diff(tt.wantErr, err); diff != "" {
 				t.Errorf("unexpected error (-want,+got):\n%s", diff)
 			}
 
-			if name != tt.wantWorkloadPriorityClassName {
-				t.Errorf("unexpected name: got: %s, expected: %s", name, tt.wantWorkloadPriorityClassName)
+			if diff := cmp.Diff(tt.wantPriorityClassRef, priorityClassRef); diff != "" {
+				t.Errorf("unexpected priortyClassRef (-want,+got):\n%s", diff)
 			}
 
-			if source != tt.wantWorkloadPriorityClassSource {
-				t.Errorf("unexpected source: got: %s, expected: %s", source, tt.wantWorkloadPriorityClassSource)
-			}
-
-			if value != tt.wantWorkloadPriorityClassValue {
-				t.Errorf("unexpected value: got: %d, expected: %d", value, tt.wantWorkloadPriorityClassValue)
+			if value != tt.wantPriorityClassValue {
+				t.Errorf("unexpected value: got: %d, expected: %d", value, tt.wantPriorityClassValue)
 			}
 		})
 	}
diff --git a/pkg/util/testing/v1beta2/wrappers.go b/pkg/util/testing/v1beta2/wrappers.go
index ff7d5a615..6c2198efd 100644
--- a/pkg/util/testing/v1beta2/wrappers.go
+++ b/pkg/util/testing/v1beta2/wrappers.go
@@ -238,11 +238,6 @@ func (w *WorkloadWrapper) Creation(t time.Time) *WorkloadWrapper {
 	return w
 }
 
-func (w *WorkloadWrapper) PriorityClass(priorityClassName string) *WorkloadWrapper {
-	w.Spec.PriorityClassName = priorityClassName
-	return w
-}
-
 func (w *WorkloadWrapper) RuntimeClass(name string) *WorkloadWrapper {
 	for i := range w.Spec.PodSets {
 		w.Spec.PodSets[i].Template.Spec.RuntimeClassName = &name
@@ -250,13 +245,21 @@ func (w *WorkloadWrapper) RuntimeClass(name string) *WorkloadWrapper {
 	return w
 }
 
-func (w *WorkloadWrapper) Priority(priority int32) *WorkloadWrapper {
-	w.Spec.Priority = &priority
+func (w *WorkloadWrapper) PriorityClassRef(ref *kueue.PriorityClassRef) *WorkloadWrapper {
+	w.Spec.PriorityClassRef = ref
 	return w
 }
 
-func (w *WorkloadWrapper) PriorityClassSource(source string) *WorkloadWrapper {
-	w.Spec.PriorityClassSource = source
+func (w *WorkloadWrapper) WorkloadPriorityClassRef(name string) *WorkloadWrapper {
+	return w.PriorityClassRef(kueue.NewWorkloadPriorityClassRef(name))
+}
+
+func (w *WorkloadWrapper) PodPriorityClassRef(name string) *WorkloadWrapper {
+	return w.PriorityClassRef(kueue.NewPodPriorityClassRef(name))
+}
+
+func (w *WorkloadWrapper) Priority(priority int32) *WorkloadWrapper {
+	w.Spec.Priority = &priority
 	return w
 }
 
diff --git a/test/integration/singlecluster/controller/core/workload_controller_test.go b/test/integration/singlecluster/controller/core/workload_controller_test.go
index 5ed2bf80e..0f2f5be83 100644
--- a/test/integration/singlecluster/controller/core/workload_controller_test.go
+++ b/test/integration/singlecluster/controller/core/workload_controller_test.go
@@ -30,7 +30,6 @@ import (
 
 	config "sigs.k8s.io/kueue/apis/config/v1beta2"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
-	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
 	"sigs.k8s.io/kueue/pkg/util/slices"
@@ -414,7 +413,9 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
 		ginkgo.It("case of WorkloadPriorityClass", func() {
 			ginkgo.By("creating workload")
 			wl = utiltestingapi.MakeWorkload("wl", ns.Name).Queue("lq").Request(corev1.ResourceCPU, "1").
-				PriorityClass("workload-priority-class").PriorityClassSource(constants.WorkloadPriorityClassSource).Priority(200).Obj()
+				WorkloadPriorityClassRef("workload-priority-class").
+				Priority(200).
+				Obj()
 			util.MustCreate(ctx, k8sClient, wl)
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &updatedQueueWorkload)).To(gomega.Succeed())
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 a04eef776..c9c0097cd 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -1394,8 +1394,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			ginkgo.By("checking that the second workload is 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(utiltesting.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
@@ -1495,8 +1493,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			ginkgo.By("checking that the second workload is 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(utiltesting.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
diff --git a/test/integration/singlecluster/webhook/core/workload_test.go b/test/integration/singlecluster/webhook/core/workload_test.go
index 9ad6638c7..86994368e 100644
--- a/test/integration/singlecluster/webhook/core/workload_test.go
+++ b/test/integration/singlecluster/webhook/core/workload_test.go
@@ -33,7 +33,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
-	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	utiltestingapi "sigs.k8s.io/kueue/pkg/util/testing/v1beta2"
@@ -168,23 +167,71 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 					).Obj()
 				},
 				utiltesting.BeInvalidError()),
-			ginkgo.Entry("invalid priorityClassName",
+			ginkgo.Entry("invalid priorityClassRef.name",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						PriorityClass("invalid_class").
+						PodPriorityClassRef("invalid_class").
 						Priority(0).
 						Obj()
 				},
 				utiltesting.BeInvalidError()),
-			ginkgo.Entry("empty priorityClassName is valid",
+			ginkgo.Entry("invalid priorityClassRef.group",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassRef(&kueue.PriorityClassRef{
+							Group: "test",
+							Kind:  kueue.PodPriorityClassKind,
+							Name:  "low",
+						}).
+						Priority(100).
+						Obj()
+				},
+				utiltesting.BeInvalidError()),
+			ginkgo.Entry("invalid priorityClassRef.kind",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassRef(&kueue.PriorityClassRef{
+							Group: kueue.PodPriorityClassGroup,
+							Kind:  "test",
+							Name:  "low",
+						}).
+						Priority(100).
+						Obj()
+				},
+				utiltesting.BeInvalidError()),
+			ginkgo.Entry("incompatible priorityClassRef.kind (scheduling.k8s.io)",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassRef(&kueue.PriorityClassRef{
+							Group: kueue.PodPriorityClassGroup,
+							Kind:  kueue.WorkloadPriorityClassKind,
+							Name:  "low",
+						}).
+						Priority(100).
+						Obj()
+				},
+				utiltesting.BeInvalidError()),
+			ginkgo.Entry("incompatible priorityClassRef.kind (kueue.x-k8s.io)",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						PriorityClassRef(&kueue.PriorityClassRef{
+							Group: kueue.WorkloadPriorityClassGroup,
+							Kind:  kueue.PodPriorityClassKind,
+							Name:  "low",
+						}).
+						Priority(100).
+						Obj()
+				},
+				utiltesting.BeInvalidError()),
+			ginkgo.Entry("omitted priorityClassRef is valid",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Priority(0).Obj()
 				},
 				gomega.Succeed()),
-			ginkgo.Entry("priority should not be nil when priorityClassName is set",
+			ginkgo.Entry("priority should not be nil when priorityClassRef is set",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						PriorityClass("priority").
+						PodPriorityClassRef("priority").
 						Obj()
 				},
 				utiltesting.BeInvalidError()),
@@ -638,34 +685,6 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				gomega.Succeed(),
 			),
-			ginkgo.Entry("priorityClassSource should not be updated",
-				func() *kueue.Workload {
-					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						Queue("q").
-						PriorityClass("test-class").PriorityClassSource(constants.PodPriorityClassSource).
-						Priority(10).
-						Obj()
-				},
-				true,
-				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = constants.WorkloadPriorityClassSource
-				},
-				utiltesting.BeInvalidError(),
-			),
-			ginkgo.Entry("priorityClassName should not be updated",
-				func() *kueue.Workload {
-					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						Queue("q").
-						PriorityClass("test-class-1").PriorityClassSource(constants.PodPriorityClassSource).
-						Priority(10).
-						Obj()
-				},
-				true,
-				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassName = "test-class-2"
-				},
-				utiltesting.BeInvalidError(),
-			),
 			ginkgo.Entry("should change other fields of admissionchecks when podSetUpdates is immutable",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
@@ -694,32 +713,6 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				gomega.Succeed(),
 			),
-			ginkgo.Entry("updating priorityClassName before setting reserve quota for workload",
-				func() *kueue.Workload {
-					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						Queue("q").
-						PriorityClass("test-class-1").PriorityClassSource(constants.PodPriorityClassSource).
-						Priority(10).Obj()
-				},
-				false,
-				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassName = "test-class-2"
-				},
-				gomega.Succeed(),
-			),
-			ginkgo.Entry("updating priorityClassSource before setting reserve quota for workload",
-				func() *kueue.Workload {
-					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						Queue("q").
-						PriorityClass("test-class").PriorityClassSource(constants.PodPriorityClassSource).
-						Priority(10).Obj()
-				},
-				false,
-				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = constants.WorkloadPriorityClassSource
-				},
-				gomega.Succeed(),
-			),
 			ginkgo.Entry("updating podSets before setting reserve quota for workload",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
@@ -747,16 +740,6 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				gomega.Succeed(),
 			),
-			ginkgo.Entry("Should allow the change of priority",
-				func() *kueue.Workload {
-					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
-				},
-				false,
-				func(newWL *kueue.Workload) {
-					newWL.Spec.Priority = ptr.To[int32](10)
-				},
-				gomega.Succeed(),
-			),
 			ginkgo.Entry("Should forbid the change of spec.podSet",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
@@ -844,14 +827,59 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				utiltesting.BeInvalidError(),
 			),
-			ginkgo.Entry("can set workload priority class when QuotaReserved=false",
+			ginkgo.Entry("Should allow to change priority when QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.Priority = ptr.To[int32](10)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("Should allow to change priority when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.Priority = ptr.To[int32](10)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("Should allow to change priority with priorityClassRef when QuotaReserved=false",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						Priority(0).
+						PodPriorityClassRef("low").
+						Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.Priority = ptr.To[int32](10)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("Should allow to change priority with priorityClassRef when QuotaReserved=true",
+				func() *kueue.Workload {
+					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
+						Priority(0).
+						PodPriorityClassRef("low").
+						Obj()
+				},
+				false,
+				func(newWL *kueue.Workload) {
+					newWL.Spec.Priority = ptr.To[int32](10)
+				},
+				gomega.Succeed(),
+			),
+			ginkgo.Entry("can set workload priorityClassRef 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.PriorityClassRef = kueue.NewWorkloadPriorityClassRef("low")
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				gomega.Succeed(),
@@ -862,8 +890,7 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				true,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = constants.WorkloadPriorityClassSource
-					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.PriorityClassRef = kueue.NewWorkloadPriorityClassRef("low")
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				utiltesting.BeInvalidError(),
@@ -871,14 +898,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			ginkgo.Entry("can update workload priority class when QuotaReserved=false",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						PriorityClassSource(constants.WorkloadPriorityClassSource).
-						PriorityClass("high").
+						WorkloadPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				false,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.PriorityClassRef.Name = "low"
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				gomega.Succeed(),
@@ -886,14 +912,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			ginkgo.Entry("can update workload priority class when QuotaReserved=true",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						PriorityClassSource(constants.WorkloadPriorityClassSource).
-						PriorityClass("high").
+						WorkloadPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				true,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.PriorityClassRef.Name = "low"
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				gomega.Succeed(),
@@ -901,15 +926,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			ginkgo.Entry("can delete workload priority class when QuotaReserved=false",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						PriorityClassSource(constants.WorkloadPriorityClassSource).
-						PriorityClass("high").
+						WorkloadPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				false,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = ""
-					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.PriorityClassRef = nil
 					newWL.Spec.Priority = nil
 				},
 				gomega.Succeed(),
@@ -917,15 +940,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			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").
+						WorkloadPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				true,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = ""
-					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.PriorityClassRef = nil
 					newWL.Spec.Priority = nil
 				},
 				utiltesting.BeInvalidError(),
@@ -936,8 +957,7 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				false,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = constants.PodPriorityClassSource
-					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.PriorityClassRef = kueue.NewPodPriorityClassRef("low")
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				gomega.Succeed(),
@@ -948,8 +968,7 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 				},
 				true,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = constants.PodPriorityClassSource
-					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.PriorityClassRef = kueue.NewPodPriorityClassRef("low")
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				utiltesting.BeInvalidError(),
@@ -957,14 +976,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			ginkgo.Entry("can update pod priority class when QuotaReserved=false",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						PriorityClassSource(constants.PodPriorityClassSource).
-						PriorityClass("high").
+						PodPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				false,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.PriorityClassRef.Name = "low"
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				gomega.Succeed(),
@@ -972,14 +990,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			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").
+						PodPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				true,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassName = "low"
+					newWL.Spec.PriorityClassRef.Name = "low"
 					newWL.Spec.Priority = ptr.To[int32](100)
 				},
 				utiltesting.BeInvalidError(),
@@ -987,15 +1004,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			ginkgo.Entry("can delete pod priority class when QuotaReserved=false",
 				func() *kueue.Workload {
 					return utiltestingapi.MakeWorkload(workloadName, ns.Name).
-						PriorityClassSource(constants.PodPriorityClassSource).
-						PriorityClass("high").
+						PodPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				false,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = ""
-					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.PriorityClassRef = nil
 					newWL.Spec.Priority = nil
 				},
 				gomega.Succeed(),
@@ -1003,15 +1018,13 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 			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").
+						PodPriorityClassRef("high").
 						Priority(1000).
 						Obj()
 				},
 				true,
 				func(newWL *kueue.Workload) {
-					newWL.Spec.PriorityClassSource = ""
-					newWL.Spec.PriorityClassName = ""
+					newWL.Spec.PriorityClassRef = nil
 					newWL.Spec.Priority = nil
 				},
 				utiltesting.BeInvalidError(),
@@ -1055,7 +1068,7 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 
 		ginkgo.It("Should have priority once priorityClassName is set", func() {
 			ginkgo.By("Creating a new Workload")
-			workload := utiltestingapi.MakeWorkload(workloadName, ns.Name).PriorityClass("priority").Obj()
+			workload := utiltestingapi.MakeWorkload(workloadName, ns.Name).PodPriorityClassRef("priority").Obj()
 			err := k8sClient.Create(ctx, workload)
 			gomega.Expect(err).Should(utiltesting.BeInvalidError())
 		})
@@ -1063,7 +1076,9 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 		ginkgo.It("workload's priority should be mutable when referencing WorkloadPriorityClass", func() {
 			ginkgo.By("creating workload")
 			wl := utiltestingapi.MakeWorkload("wl", ns.Name).Queue("lq").Request(corev1.ResourceCPU, "1").
-				PriorityClass("workload-priority-class").PriorityClassSource(constants.WorkloadPriorityClassSource).Priority(200).Obj()
+				WorkloadPriorityClassRef("workload-priority-class").
+				Priority(200).
+				Obj()
 			util.MustCreate(ctx, k8sClient, wl)
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &updatedQueueWorkload)).To(gomega.Succeed())
@@ -1085,7 +1100,9 @@ var _ = ginkgo.Describe("Workload validating webhook", ginkgo.Ordered, func() {
 		ginkgo.It("workload's priority should be mutable when referencing PriorityClass", func() {
 			ginkgo.By("creating workload")
 			wl := utiltestingapi.MakeWorkload("wl", ns.Name).Queue("lq").Request(corev1.ResourceCPU, "1").
-				PriorityClass("priority-class").PriorityClassSource(constants.PodPriorityClassSource).Priority(100).Obj()
+				PodPriorityClassRef("priority-class").
+				Priority(100).
+				Obj()
 			util.MustCreate(ctx, k8sClient, wl)
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &updatedQueueWorkload)).To(gomega.Succeed())
diff --git a/test/util/util.go b/test/util/util.go
index 981eeb10d..68ac9828b 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -66,7 +66,6 @@ import (
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
-	"sigs.k8s.io/kueue/pkg/constants"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/scheduler/preemption"
@@ -369,22 +368,24 @@ func expectWorkloadsToBeAdmittedCountWithOffset(ctx context.Context, offset int,
 
 func ExpectWorkloadsWithWorkloadPriority(ctx context.Context, c client.Client, name string, value int32, wlKeys ...client.ObjectKey) {
 	ginkgo.GinkgoHelper()
-	expectWorkloadsWithPriority(ctx, c, constants.WorkloadPriorityClassSource, name, value, wlKeys...)
+	expectWorkloadsWithPriority(ctx, c, kueue.WorkloadPriorityClassGroup, kueue.WorkloadPriorityClassKind, name, value, wlKeys...)
 }
 
 func ExpectWorkloadsWithPodPriority(ctx context.Context, c client.Client, name string, value int32, wlKeys ...client.ObjectKey) {
 	ginkgo.GinkgoHelper()
-	expectWorkloadsWithPriority(ctx, c, constants.PodPriorityClassSource, name, value, wlKeys...)
+	expectWorkloadsWithPriority(ctx, c, kueue.PodPriorityClassGroup, kueue.PodPriorityClassKind, name, value, wlKeys...)
 }
 
-func expectWorkloadsWithPriority(ctx context.Context, c client.Client, priorityClassSource, name string, value int32, wlKeys ...client.ObjectKey) {
+func expectWorkloadsWithPriority(ctx context.Context, c client.Client, priorityClassGroup kueue.PriorityClassGroup, priorityClassKind kueue.PriorityClassKind, name string, value int32, wlKeys ...client.ObjectKey) {
 	ginkgo.GinkgoHelper()
 	createdWl := &kueue.Workload{}
 	gomega.Eventually(func(g gomega.Gomega) {
 		for _, wlKey := range wlKeys {
 			g.Expect(c.Get(ctx, wlKey, createdWl)).To(gomega.Succeed())
-			g.Expect(createdWl.Spec.PriorityClassSource).To(gomega.Equal(priorityClassSource))
-			g.Expect(createdWl.Spec.PriorityClassName).To(gomega.Equal(name))
+			g.Expect(createdWl.Spec.PriorityClassRef).ToNot(gomega.BeNil())
+			g.Expect(createdWl.Spec.PriorityClassRef.Group).To(gomega.Equal(priorityClassGroup))
+			g.Expect(createdWl.Spec.PriorityClassRef.Kind).To(gomega.Equal(priorityClassKind))
+			g.Expect(createdWl.Spec.PriorityClassRef.Name).To(gomega.Equal(name))
 			g.Expect(createdWl.Spec.Priority).To(gomega.Equal(&value))
 		}
 	}, Timeout, Interval).Should(gomega.Succeed())