Repository: kubernetes-sigs/kueue

Issue:
Consider dropping IsManagingObjectsOwner as it seems redundant
<!-- Please only use this template for submitting clean up requests -->

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

Consider dropping IsManagingObjectsOwner as it seems redundant. 

**Why is this needed**:

To simplify code. All Jobs implement it just by checking GVK.

It was discussed a little bit on https://github.com/kubernetes-sigs/kueue/pull/4808#discussion_r2039537827

Discussion:
@mimowo: cc @mbobrovskyi @mbobrovskyi: /assign

Solution patch:
diff --git a/Makefile-test.mk b/Makefile-test.mk
index c463984aa..67d8b4be0 100644
--- a/Makefile-test.mk
+++ b/Makefile-test.mk
@@ -177,6 +177,7 @@ run-test-e2e-customconfigs-%: FORCE
 		ARTIFACTS="$(ARTIFACTS)/$@" IMAGE_TAG=$(IMAGE_TAG) GINKGO_ARGS="$(GINKGO_ARGS)" \
 		KIND_CLUSTER_FILE="kind-cluster.yaml" E2E_TARGET_FOLDER="customconfigs" \
 		JOBSET_VERSION=$(JOBSET_VERSION) APPWRAPPER_VERSION=$(APPWRAPPER_VERSION) \
+		LEADERWORKERSET_VERSION=$(LEADERWORKERSET_VERSION) \
 		TEST_LOG_LEVEL=$(TEST_LOG_LEVEL) \
 		./hack/e2e-test.sh
 	$(PROJECT_DIR)/bin/ginkgo-top -i $(ARTIFACTS)/$@/e2e.json > $(ARTIFACTS)/$@/e2e-top.yaml
diff --git a/charts/kueue/templates/rbac/role.yaml b/charts/kueue/templates/rbac/role.yaml
index 947ebd37d..0440eb8d2 100644
--- a/charts/kueue/templates/rbac/role.yaml
+++ b/charts/kueue/templates/rbac/role.yaml
@@ -82,6 +82,7 @@ rules:
   - apiGroups:
       - apps
     resources:
+      - deployments
       - replicasets
       - statefulsets
     verbs:
diff --git a/config/components/rbac/role.yaml b/config/components/rbac/role.yaml
index 8ca7211e8..642e4fa5d 100644
--- a/config/components/rbac/role.yaml
+++ b/config/components/rbac/role.yaml
@@ -81,6 +81,7 @@ rules:
 - apiGroups:
   - apps
   resources:
+  - deployments
   - replicasets
   - statefulsets
   verbs:
diff --git a/pkg/controller/jobframework/defaults.go b/pkg/controller/jobframework/defaults.go
index e4c20ac03..69bee70bd 100644
--- a/pkg/controller/jobframework/defaults.go
+++ b/pkg/controller/jobframework/defaults.go
@@ -45,14 +45,13 @@ func ApplyDefaultForSuspend(ctx context.Context, job GenericJob, k8sClient clien
 	return nil
 }
 
-// +kubebuilder:rbac:groups="apps",resources=replicasets,verbs=get;list;watch
-
 // WorkloadShouldBeSuspended determines whether jobObj should be default suspended on creation
 func WorkloadShouldBeSuspended(ctx context.Context, jobObj client.Object, k8sClient client.Client,
 	manageJobsWithoutQueueName bool, managedJobsNamespaceSelector labels.Selector) (bool, error) {
-	// Do not default suspend a job whose owner is already managed by Kueue
-	if IsOwnerManagedByKueueForObject(jobObj) {
-		return false, nil
+	// Do not default suspend a job whose ancestor is already managed by Kueue
+	ancestorJob, err := FindAncestorJobManagedByKueue(ctx, k8sClient, jobObj, manageJobsWithoutQueueName)
+	if err != nil || ancestorJob != nil {
+		return false, err
 	}
 
 	// Jobs with queue names whose parents are not managed by Kueue are default suspended
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index 3089edfbc..3f7873f8c 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -67,11 +67,12 @@ const (
 )
 
 var (
-	ErrCyclicOwnership          = errors.New("cyclic ownership")
-	ErrWorkloadOwnerNotFound    = errors.New("workload owner not found")
-	ErrNoMatchingWorkloads      = errors.New("no matching workloads")
-	ErrExtraWorkloads           = errors.New("extra workloads")
-	ErrPrebuiltWorkloadNotFound = errors.New("prebuilt workload not found")
+	ErrCyclicOwnership                = errors.New("cyclic ownership")
+	ErrWorkloadOwnerNotFound          = errors.New("workload owner not found")
+	ErrManagedOwnersChainLimitReached = errors.New("managed owner chain limit reached")
+	ErrNoMatchingWorkloads            = errors.New("no matching workloads")
+	ErrExtraWorkloads                 = errors.New("extra workloads")
+	ErrPrebuiltWorkloadNotFound       = errors.New("prebuilt workload not found")
 )
 
 // JobReconciler reconciles a GenericJob object
@@ -301,8 +302,13 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 		// Skipping traversal to top-level ancestor job because this is already a top-level job.
 		isTopLevelJob = true
 	} else {
-		ancestorJob, err = FindAncestorJobManagedByKueue(ctx, r.client, r.record, object, r.manageJobsWithoutQueueName)
+		ancestorJob, err = FindAncestorJobManagedByKueue(ctx, r.client, object, r.manageJobsWithoutQueueName)
 		if err != nil {
+			if errors.Is(err, ErrManagedOwnersChainLimitReached) {
+				errMsg := fmt.Sprintf("Terminated search for Kueue-managed Job because ancestor depth exceeded limit of %d", managedOwnersChainLimit)
+				r.record.Eventf(object, corev1.EventTypeWarning, ReasonJobNestingTooDeep, errMsg)
+				log.Error(err, errMsg)
+			}
 			return ctrl.Result{}, err
 		}
 		isTopLevelJob = ancestorJob == nil
@@ -613,7 +619,7 @@ func (r *JobReconciler) getWorkloadForObject(ctx context.Context, jobObj client.
 // Job (queue-name) -> JobSet -> AppWrapper (queue-name) => AppWrapper
 // Job (queue-name) -> JobSet (queue-name) -> AppWrapper (queue-name) => AppWrapper
 // Job -> JobSet (disabled) -> AppWrapper => AppWrapper
-func FindAncestorJobManagedByKueue(ctx context.Context, c client.Client, record record.EventRecorder, jobObj client.Object, manageJobsWithoutQueueName bool) (client.Object, error) {
+func FindAncestorJobManagedByKueue(ctx context.Context, c client.Client, jobObj client.Object, manageJobsWithoutQueueName bool) (client.Object, error) {
 	log := ctrl.LoggerFrom(ctx)
 	seen := sets.New[types.UID]()
 	currentObj := jobObj
@@ -652,15 +658,7 @@ func FindAncestorJobManagedByKueue(ctx context.Context, c client.Client, record
 		}
 		currentObj = parentObj
 		if len(seen) > managedOwnersChainLimit {
-			record.Eventf(jobObj, corev1.EventTypeWarning, ReasonJobNestingTooDeep,
-				"Terminated search for Kueue-managed Job because ancestor depth exceeded limit of %d", managedOwnersChainLimit,
-			)
-			log.V(2).Info(
-				"WARNING: Terminated search for Kueue-managed Job because ancestor depth exceeded managedOwnersChainLimit",
-				"limit ", managedOwnersChainLimit,
-				"lastParentReached", parentObj,
-			)
-			return topLevelJob, nil
+			return nil, ErrManagedOwnersChainLimitReached
 		}
 	}
 }
diff --git a/pkg/controller/jobframework/validation.go b/pkg/controller/jobframework/validation.go
index 84b6f2420..e6664bef4 100644
--- a/pkg/controller/jobframework/validation.go
+++ b/pkg/controller/jobframework/validation.go
@@ -204,13 +204,3 @@ func validateImmutablePodGroupPodSpecPath(newShape, oldShape map[string]any, fie
 
 	return allErrs
 }
-
-func IsManagedByKueue(obj client.Object) bool {
-	if IsOwnerManagedByKueueForObject(obj) {
-		return false
-	}
-	if QueueNameForObject(obj) != "" {
-		return true
-	}
-	return false
-}
diff --git a/pkg/controller/jobs/deployment/deployment_controller.go b/pkg/controller/jobs/deployment/deployment_controller.go
index 42382df75..7cf79b819 100644
--- a/pkg/controller/jobs/deployment/deployment_controller.go
+++ b/pkg/controller/jobs/deployment/deployment_controller.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -38,18 +39,26 @@ const (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:   SetupIndexes,
-		NewReconciler:  jobframework.NewNoopReconcilerFactory(gvk),
-		GVK:            gvk,
-		SetupWebhook:   SetupWebhook,
-		JobType:        &appsv1.Deployment{},
-		AddToScheme:    appsv1.AddToScheme,
-		DependencyList: []string{"pod"},
+		SetupIndexes:           SetupIndexes,
+		NewReconciler:          jobframework.NewNoopReconcilerFactory(gvk),
+		GVK:                    gvk,
+		SetupWebhook:           SetupWebhook,
+		JobType:                &appsv1.Deployment{},
+		AddToScheme:            appsv1.AddToScheme,
+		DependencyList:         []string{"pod"},
+		IsManagingObjectsOwner: isDeployment,
 	}))
 }
 
+func isDeployment(ref *metav1.OwnerReference) bool {
+	return ref.Kind == "Deployment" && ref.APIVersion == "apps/v1"
+}
+
 type Deployment appsv1.Deployment
 
+// +kubebuilder:rbac:groups="apps",resources=deployments,verbs=get;list;watch
+// +kubebuilder:rbac:groups="apps",resources=replicasets,verbs=get;list;watch
+
 func fromObject(o runtime.Object) *Deployment {
 	return (*Deployment)(o.(*appsv1.Deployment))
 }
diff --git a/pkg/controller/jobs/deployment/deployment_webhook.go b/pkg/controller/jobs/deployment/deployment_webhook.go
index c8d265fd8..bcc85697d 100644
--- a/pkg/controller/jobs/deployment/deployment_webhook.go
+++ b/pkg/controller/jobs/deployment/deployment_webhook.go
@@ -28,7 +28,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
-	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/constants"
+	controllerconstants "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework/webhook"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
@@ -81,12 +82,13 @@ func (wh *Webhook) Default(ctx context.Context, obj runtime.Object) error {
 		if deployment.Spec.Template.Labels == nil {
 			deployment.Spec.Template.Labels = make(map[string]string, 1)
 		}
+		deployment.Spec.Template.Labels[constants.ManagedByKueueLabelKey] = constants.ManagedByKueueLabelValue
 		queueName := jobframework.QueueNameForObject(deployment.Object())
 		if queueName != "" {
-			deployment.Spec.Template.Labels[constants.QueueLabel] = queueName
+			deployment.Spec.Template.Labels[controllerconstants.QueueLabel] = queueName
 		}
 		if priorityClass := jobframework.WorkloadPriorityClassName(deployment.Object()); priorityClass != "" {
-			deployment.Spec.Template.Labels[constants.WorkloadPriorityClassLabel] = priorityClass
+			deployment.Spec.Template.Labels[controllerconstants.WorkloadPriorityClassLabel] = priorityClass
 		}
 	}
 
@@ -110,7 +112,7 @@ func (wh *Webhook) ValidateCreate(ctx context.Context, obj runtime.Object) (warn
 
 var (
 	labelsPath         = field.NewPath("metadata", "labels")
-	queueNameLabelPath = labelsPath.Key(constants.QueueLabel)
+	queueNameLabelPath = labelsPath.Key(controllerconstants.QueueLabel)
 )
 
 func (wh *Webhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (warnings admission.Warnings, err error) {
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
index 907b74c68..dd2bd94a5 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
@@ -33,7 +33,8 @@ import (
 	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
-	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/constants"
+	controllerconstants "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	clientutil "sigs.k8s.io/kueue/pkg/util/client"
@@ -101,7 +102,7 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 
 func (r *PodReconciler) setDefault(ctx context.Context, pod *corev1.Pod) (bool, error) {
 	// If queue label already exist nothing to update.
-	if _, ok := pod.Labels[constants.QueueLabel]; ok {
+	if _, ok := pod.Labels[controllerconstants.QueueLabel]; ok {
 		return false, nil
 	}
 
@@ -124,9 +125,10 @@ func (r *PodReconciler) setDefault(ctx context.Context, pod *corev1.Pod) (bool,
 
 	wlName := GetWorkloadName(lws.UID, lws.Name, pod.Labels[leaderworkersetv1.GroupIndexLabelKey])
 
-	pod.Labels[constants.QueueLabel] = queueName
+	pod.Labels[constants.ManagedByKueueLabelKey] = constants.ManagedByKueueLabelValue
+	pod.Labels[controllerconstants.QueueLabel] = queueName
 	pod.Labels[podconstants.GroupNameLabel] = wlName
-	pod.Labels[constants.PrebuiltWorkloadLabel] = wlName
+	pod.Labels[controllerconstants.PrebuiltWorkloadLabel] = wlName
 	pod.Annotations[podconstants.GroupTotalCountAnnotation] = fmt.Sprint(ptr.Deref(lws.Spec.LeaderWorkerTemplate.Size, 1))
 	pod.Annotations[podconstants.RoleHashAnnotation] = string(kueue.DefaultPodSetName)
 
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
index 8bce859b8..d39814fef 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
@@ -20,8 +20,11 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -43,18 +46,24 @@ const (
 )
 
 type Reconciler struct {
-	client          client.Client
-	record          record.EventRecorder
-	labelKeysToCopy []string
+	client                       client.Client
+	log                          logr.Logger
+	record                       record.EventRecorder
+	labelKeysToCopy              []string
+	manageJobsWithoutQueueName   bool
+	managedJobsNamespaceSelector labels.Selector
 }
 
 func NewReconciler(client client.Client, eventRecorder record.EventRecorder, opts ...jobframework.Option) jobframework.JobReconcilerInterface {
 	options := jobframework.ProcessOptions(opts...)
 
 	return &Reconciler{
-		client:          client,
-		record:          eventRecorder,
-		labelKeysToCopy: options.LabelKeysToCopy,
+		client:                       client,
+		log:                          ctrl.Log.WithName("leaderworkerset-reconciler"),
+		record:                       eventRecorder,
+		labelKeysToCopy:              options.LabelKeysToCopy,
+		manageJobsWithoutQueueName:   options.ManageJobsWithoutQueueName,
+		managedJobsNamespaceSelector: options.ManagedJobsNamespaceSelector,
 	}
 }
 
@@ -212,6 +221,16 @@ func (r *Reconciler) handle(obj client.Object) bool {
 	if !isLws {
 		return false
 	}
+
+	ctx := context.Background()
+	log := r.log.WithValues("leaderworkerset", klog.KObj(lws))
+	ctrl.LoggerInto(ctx, log)
+
 	// Handle only leaderworkerset managed by kueue.
-	return jobframework.IsManagedByKueue(lws)
+	suspend, err := jobframework.WorkloadShouldBeSuspended(ctx, lws, r.client, r.manageJobsWithoutQueueName, r.managedJobsNamespaceSelector)
+	if err != nil {
+		log.Error(err, "Failed to determine if the LeaderWorkerSet should be managed by Kueue")
+	}
+
+	return suspend
 }
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
index 0c3a2a0ea..bfcb0ba2d 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_webhook.go
@@ -45,9 +45,13 @@ type Webhook struct {
 	queues                       *queue.Manager
 }
 
-func SetupWebhook(mgr ctrl.Manager, _ ...jobframework.Option) error {
+func SetupWebhook(mgr ctrl.Manager, opts ...jobframework.Option) error {
+	options := jobframework.ProcessOptions(opts...)
 	wh := &Webhook{
-		client: mgr.GetClient(),
+		client:                       mgr.GetClient(),
+		manageJobsWithoutQueueName:   options.ManageJobsWithoutQueueName,
+		managedJobsNamespaceSelector: options.ManagedJobsNamespaceSelector,
+		queues:                       options.Queues,
 	}
 	return ctrl.NewWebhookManagedBy(mgr).
 		For(&leaderworkersetv1.LeaderWorkerSet{}).
@@ -141,7 +145,11 @@ func (wh *Webhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Ob
 		oldLeaderWorkerSet.Object(),
 	)...)
 
-	if jobframework.IsManagedByKueue(newLeaderWorkerSet.Object()) {
+	suspend, err := jobframework.WorkloadShouldBeSuspended(ctx, newLeaderWorkerSet.Object(), wh.client, wh.manageJobsWithoutQueueName, wh.managedJobsNamespaceSelector)
+	if err != nil {
+		return nil, err
+	}
+	if suspend {
 		allErrs = append(allErrs, validateImmutablePodTemplateSpec(
 			newLeaderWorkerSet.Spec.LeaderWorkerTemplate.LeaderTemplate,
 			oldLeaderWorkerSet.Spec.LeaderWorkerTemplate.LeaderTemplate,
diff --git a/pkg/controller/jobs/pod/pod_controller.go b/pkg/controller/jobs/pod/pod_controller.go
index 34c25c7b4..941c0bf49 100644
--- a/pkg/controller/jobs/pod/pod_controller.go
+++ b/pkg/controller/jobs/pod/pod_controller.go
@@ -90,15 +90,20 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:      SetupIndexes,
-		NewJob:            NewJob,
-		NewReconciler:     NewReconciler,
-		SetupWebhook:      SetupWebhook,
-		JobType:           &corev1.Pod{},
-		MultiKueueAdapter: &multiKueueAdapter{},
+		SetupIndexes:           SetupIndexes,
+		NewJob:                 NewJob,
+		NewReconciler:          NewReconciler,
+		SetupWebhook:           SetupWebhook,
+		JobType:                &corev1.Pod{},
+		MultiKueueAdapter:      &multiKueueAdapter{},
+		IsManagingObjectsOwner: isPod,
 	}))
 }
 
+func isPod(ref *metav1.OwnerReference) bool {
+	return ref.Kind == "Pod" && ref.APIVersion == "v1"
+}
+
 // +kubebuilder:rbac:groups=scheduling.k8s.io,resources=priorityclasses,verbs=list;get;watch
 // +kubebuilder:rbac:groups="",resources=events,verbs=create;watch;update;patch
 // +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;update;patch;delete
diff --git a/pkg/controller/jobs/pod/pod_webhook.go b/pkg/controller/jobs/pod/pod_webhook.go
index fd2150078..3e5d6615f 100644
--- a/pkg/controller/jobs/pod/pod_webhook.go
+++ b/pkg/controller/jobs/pod/pod_webhook.go
@@ -21,7 +21,6 @@ import (
 	"errors"
 	"fmt"
 
-	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -164,19 +163,9 @@ func (w *PodWebhook) Default(ctx context.Context, obj runtime.Object) error {
 		}
 
 		// Do not suspend a Pod whose owner is already managed by Kueue
-		if owner := metav1.GetControllerOf(pod.Object()); owner != nil {
-			if owner.Kind == "ReplicaSet" && owner.APIVersion == "apps/v1" {
-				// ReplicaSet is an implementation detail; skip over it to the user-facing framework
-				rs := &appsv1.ReplicaSet{}
-				err := w.client.Get(ctx, client.ObjectKey{Name: owner.Name, Namespace: pod.pod.GetNamespace()}, rs)
-				if err != nil {
-					return fmt.Errorf("failed to get replicaset: %w", err)
-				}
-				owner = metav1.GetControllerOf(rs)
-			}
-			if owner != nil && jobframework.IsOwnerIntegrationEnabled(owner) {
-				return nil
-			}
+		ancestorJob, err := jobframework.FindAncestorJobManagedByKueue(ctx, w.client, pod.Object(), w.manageJobsWithoutQueueName)
+		if err != nil || ancestorJob != nil {
+			return err
 		}
 
 		// Local queue defaulting
@@ -190,20 +179,23 @@ func (w *PodWebhook) Default(ctx context.Context, obj runtime.Object) error {
 		}
 
 		suspend = jobframework.QueueNameForObject(pod.Object()) != "" || w.manageJobsWithoutQueueName
+		if suspend {
+			if pod.pod.Labels == nil {
+				pod.pod.Labels = make(map[string]string)
+			}
+			pod.pod.Labels[constants.ManagedByKueueLabelKey] = constants.ManagedByKueueLabelValue
+		}
 	}
 
 	if suspend {
 		controllerutil.AddFinalizer(pod.Object(), podconstants.PodFinalizer)
-
-		if pod.pod.Labels == nil {
-			pod.pod.Labels = make(map[string]string)
-		}
-		pod.pod.Labels[constants.ManagedByKueueLabelKey] = constants.ManagedByKueueLabelValue
-
 		gate(&pod.pod)
 
 		if features.Enabled(features.TopologyAwareScheduling) {
 			if val, ok := pod.pod.Annotations[kueuealpha.PodGroupPodIndexLabelAnnotation]; ok {
+				if pod.pod.Labels == nil {
+					pod.pod.Labels = make(map[string]string, 1)
+				}
 				pod.pod.Labels[kueuealpha.PodGroupPodIndexLabel] = pod.pod.Labels[val]
 			}
 			utilpod.Gate(&pod.pod, kueuealpha.TopologySchedulingGate)
diff --git a/pkg/controller/jobs/statefulset/statefulset_controller.go b/pkg/controller/jobs/statefulset/statefulset_controller.go
index 0039a747c..502e922dc 100644
--- a/pkg/controller/jobs/statefulset/statefulset_controller.go
+++ b/pkg/controller/jobs/statefulset/statefulset_controller.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -38,16 +39,21 @@ const (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:   SetupIndexes,
-		NewReconciler:  NewReconciler,
-		SetupWebhook:   SetupWebhook,
-		JobType:        &appsv1.StatefulSet{},
-		AddToScheme:    appsv1.AddToScheme,
-		DependencyList: []string{"pod"},
-		GVK:            gvk,
+		SetupIndexes:           SetupIndexes,
+		NewReconciler:          NewReconciler,
+		SetupWebhook:           SetupWebhook,
+		JobType:                &appsv1.StatefulSet{},
+		AddToScheme:            appsv1.AddToScheme,
+		DependencyList:         []string{"pod"},
+		GVK:                    gvk,
+		IsManagingObjectsOwner: isStatefulSet,
 	}))
 }
 
+func isStatefulSet(ref *metav1.OwnerReference) bool {
+	return ref.Kind == "StatefulSet" && ref.APIVersion == "apps/v1"
+}
+
 type StatefulSet appsv1.StatefulSet
 
 func fromObject(o runtime.Object) *StatefulSet {
diff --git a/pkg/controller/jobs/statefulset/statefulset_reconciler.go b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
index c8b2464c7..9274cca5f 100644
--- a/pkg/controller/jobs/statefulset/statefulset_reconciler.go
+++ b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
@@ -20,9 +20,11 @@ import (
 	"context"
 	"time"
 
+	"github.com/go-logr/logr"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
@@ -53,7 +55,10 @@ var (
 )
 
 type Reconciler struct {
-	client client.Client
+	client                       client.Client
+	log                          logr.Logger
+	manageJobsWithoutQueueName   bool
+	managedJobsNamespaceSelector labels.Selector
 }
 
 func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
@@ -143,8 +148,15 @@ func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Complete(r)
 }
 
-func NewReconciler(client client.Client, _ record.EventRecorder, _ ...jobframework.Option) jobframework.JobReconcilerInterface {
-	return &Reconciler{client: client}
+func NewReconciler(client client.Client, _ record.EventRecorder, opts ...jobframework.Option) jobframework.JobReconcilerInterface {
+	options := jobframework.ProcessOptions(opts...)
+
+	return &Reconciler{
+		client:                       client,
+		log:                          ctrl.Log.WithName("statefulset-reconciler"),
+		manageJobsWithoutQueueName:   options.ManageJobsWithoutQueueName,
+		managedJobsNamespaceSelector: options.ManagedJobsNamespaceSelector,
+	}
 }
 
 var _ predicate.Predicate = (*Reconciler)(nil)
@@ -170,8 +182,18 @@ func (r *Reconciler) handle(obj client.Object) bool {
 	if !isSts {
 		return true
 	}
+
+	ctx := context.Background()
+	log := r.log.WithValues("statefulset", klog.KObj(sts))
+	ctrl.LoggerInto(ctx, log)
+
 	// Handle only statefulset managed by kueue.
-	return jobframework.IsManagedByKueue(sts)
+	suspend, err := jobframework.WorkloadShouldBeSuspended(ctx, sts, r.client, r.manageJobsWithoutQueueName, r.managedJobsNamespaceSelector)
+	if err != nil {
+		log.Error(err, "Failed to determine if the StatefulSet should be managed by Kueue")
+	}
+
+	return suspend
 }
 
 var _ handler.EventHandler = (*podHandler)(nil)
diff --git a/pkg/controller/jobs/statefulset/statefulset_webhook.go b/pkg/controller/jobs/statefulset/statefulset_webhook.go
index e48939645..f23478e86 100644
--- a/pkg/controller/jobs/statefulset/statefulset_webhook.go
+++ b/pkg/controller/jobs/statefulset/statefulset_webhook.go
@@ -32,7 +32,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
-	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/constants"
+	controllerconstants "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/queue"
@@ -79,12 +80,13 @@ func (wh *Webhook) Default(ctx context.Context, obj runtime.Object) error {
 			ss.Spec.Template.Annotations = make(map[string]string, 1)
 		}
 		ss.Spec.Template.Annotations[podconstants.SuspendedByParentAnnotation] = FrameworkName
+		if ss.Spec.Template.Labels == nil {
+			ss.Spec.Template.Labels = make(map[string]string, 1)
+		}
+		ss.Spec.Template.Labels[constants.ManagedByKueueLabelKey] = constants.ManagedByKueueLabelValue
 		queueName := jobframework.QueueNameForObject(ss.Object())
 		if queueName != "" {
-			if ss.Spec.Template.Labels == nil {
-				ss.Spec.Template.Labels = make(map[string]string, 2)
-			}
-			ss.Spec.Template.Labels[constants.QueueLabel] = queueName
+			ss.Spec.Template.Labels[controllerconstants.QueueLabel] = queueName
 			ss.Spec.Template.Labels[podconstants.GroupNameLabel] = GetWorkloadName(ss.Name)
 			ss.Spec.Template.Annotations[podconstants.GroupTotalCountAnnotation] = fmt.Sprint(ptr.Deref(ss.Spec.Replicas, 1))
 			ss.Spec.Template.Annotations[podconstants.GroupFastAdmissionAnnotationKey] = podconstants.GroupFastAdmissionAnnotationValue
@@ -92,7 +94,7 @@ func (wh *Webhook) Default(ctx context.Context, obj runtime.Object) error {
 			ss.Spec.Template.Annotations[kueuealpha.PodGroupPodIndexLabelAnnotation] = appsv1.PodIndexLabel
 		}
 		if priorityClass := jobframework.WorkloadPriorityClassName(ss.Object()); priorityClass != "" {
-			ss.Spec.Template.Labels[constants.WorkloadPriorityClassLabel] = priorityClass
+			ss.Spec.Template.Labels[controllerconstants.WorkloadPriorityClassLabel] = priorityClass
 		}
 	}
 
@@ -116,8 +118,8 @@ func (wh *Webhook) ValidateCreate(ctx context.Context, obj runtime.Object) (warn
 
 var (
 	labelsPath                 = field.NewPath("metadata", "labels")
-	queueNameLabelPath         = labelsPath.Key(constants.QueueLabel)
-	priorityClassNameLabelPath = labelsPath.Key(constants.WorkloadPriorityClassLabel)
+	queueNameLabelPath         = labelsPath.Key(controllerconstants.QueueLabel)
+	priorityClassNameLabelPath = labelsPath.Key(controllerconstants.WorkloadPriorityClassLabel)
 	specPath                   = field.NewPath("spec")
 	replicasPath               = specPath.Child("replicas")
 	specTemplatePath           = specPath.Child("template")
@@ -146,7 +148,11 @@ func (wh *Webhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Ob
 		newStatefulSet.Object(),
 	)...)
 
-	if jobframework.IsManagedByKueue(newStatefulSet.Object()) {
+	suspend, err := jobframework.WorkloadShouldBeSuspended(ctx, newStatefulSet.Object(), wh.client, wh.manageJobsWithoutQueueName, wh.managedJobsNamespaceSelector)
+	if err != nil {
+		return nil, err
+	}
+	if suspend {
 		allErrs = append(allErrs, jobframework.ValidateImmutablePodGroupPodSpec(
 			&newStatefulSet.Spec.Template.Spec,
 			&oldStatefulSet.Spec.Template.Spec,



Test patch:
diff --git a/pkg/controller/jobframework/defaults_test.go b/pkg/controller/jobframework/defaults_test.go
index 3ce0d8168..e9e8b4b4f 100644
--- a/pkg/controller/jobframework/defaults_test.go
+++ b/pkg/controller/jobframework/defaults_test.go
@@ -33,7 +33,7 @@ func TestWorkloadShouldBeSuspended(t *testing.T) {
 	t.Cleanup(EnableIntegrationsForTest(t, "batch/job"))
 	managedNamespace := utiltesting.MakeNamespaceWrapper("managed-ns").Label(corev1.LabelMetadataName, "managed-ns").Obj()
 	unmanagedNamespace := utiltesting.MakeNamespaceWrapper("unmanaged-ns").Label(corev1.LabelMetadataName, "unmanaged-ns").Obj()
-	parent := utiltestingjob.MakeJob("parent", managedNamespace.Name).Queue("default").Obj()
+	parent := utiltestingjob.MakeJob("parent", managedNamespace.Name).UID("parent").Queue("default").Obj()
 	ls := &metav1.LabelSelector{
 		MatchExpressions: []metav1.LabelSelectorRequirement{
 			{
diff --git a/pkg/controller/jobframework/reconciler_test.go b/pkg/controller/jobframework/reconciler_test.go
index 3c8973df5..9b09c951d 100644
--- a/pkg/controller/jobframework/reconciler_test.go
+++ b/pkg/controller/jobframework/reconciler_test.go
@@ -163,15 +163,7 @@ func TestFindAncestorJobManagedByKueue(t *testing.T) {
 			job: testingjob.MakeJob(childJobName, jobNamespace).
 				OwnerReference("ancestor-11", batchv1.SchemeGroupVersion.WithKind("Job")).
 				Obj(),
-			wantManaged: nil,
-			wantEvents: []utiltesting.EventRecord{
-				{
-					Key:       types.NamespacedName{Namespace: jobNamespace, Name: childJobName},
-					EventType: corev1.EventTypeWarning,
-					Reason:    ReasonJobNestingTooDeep,
-					Message:   "Terminated search for Kueue-managed Job because ancestor depth exceeded limit of 10",
-				},
-			},
+			wantErr: ErrManagedOwnersChainLimitReached,
 		},
 		"Job -> JobSet -> AppWrapper => nil": {
 			integrations: []string{"jobset.x-k8s.io/jobset", "workload.codeflare.dev/appwrapper"},
@@ -335,7 +327,7 @@ func TestFindAncestorJobManagedByKueue(t *testing.T) {
 				builder = builder.WithObjects(tc.job)
 			}
 			cl := builder.Build()
-			gotManaged, gotErr := FindAncestorJobManagedByKueue(ctx, cl, recorder, tc.job, tc.manageJobsWithoutQueueName)
+			gotManaged, gotErr := FindAncestorJobManagedByKueue(ctx, cl, tc.job, tc.manageJobsWithoutQueueName)
 			if diff := cmp.Diff(tc.wantManaged, gotManaged, cmp.Options{
 				cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
 				cmpopts.EquateEmpty(),
diff --git a/pkg/controller/jobs/deployment/deployment_webhook_test.go b/pkg/controller/jobs/deployment/deployment_webhook_test.go
index 4ef5a992d..b56259f92 100644
--- a/pkg/controller/jobs/deployment/deployment_webhook_test.go
+++ b/pkg/controller/jobs/deployment/deployment_webhook_test.go
@@ -51,6 +51,7 @@ func TestDefault(t *testing.T) {
 				Queue("test-queue").
 				Obj(),
 			want: testingdeployment.MakeDeployment("test-pod", "").
+				PodTemplateSpecManagedByKueue().
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
@@ -62,6 +63,7 @@ func TestDefault(t *testing.T) {
 				PodTemplateSpecQueue("test-queue").
 				Obj(),
 			want: testingdeployment.MakeDeployment("test-pod", "").
+				PodTemplateSpecManagedByKueue().
 				Queue("new-test-queue").
 				PodTemplateSpecQueue("new-test-queue").
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
@@ -76,6 +78,7 @@ func TestDefault(t *testing.T) {
 			defaultLqExist:       true,
 			deployment:           testingdeployment.MakeDeployment("test-pod", "default").Obj(),
 			want: testingdeployment.MakeDeployment("test-pod", "default").
+				PodTemplateSpecManagedByKueue().
 				Queue("default").
 				PodTemplateSpecQueue("default").
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
@@ -86,6 +89,7 @@ func TestDefault(t *testing.T) {
 			defaultLqExist:       true,
 			deployment:           testingdeployment.MakeDeployment("test-pod", "").Queue("test-queue").Obj(),
 			want: testingdeployment.MakeDeployment("test-pod", "").
+				PodTemplateSpecManagedByKueue().
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
@@ -104,6 +108,7 @@ func TestDefault(t *testing.T) {
 				Label(constants.WorkloadPriorityClassLabel, "test").
 				Obj(),
 			want: testingdeployment.MakeDeployment("test-pod", "").
+				PodTemplateSpecManagedByKueue().
 				Queue("test-queue").
 				Label(constants.WorkloadPriorityClassLabel, "test").
 				PodTemplateSpecQueue("test-queue").
@@ -119,6 +124,7 @@ func TestDefault(t *testing.T) {
 				PodTemplateSpecLabel(constants.WorkloadPriorityClassLabel, "test").
 				Obj(),
 			want: testingdeployment.MakeDeployment("test-pod", "").
+				PodTemplateSpecManagedByKueue().
 				Queue("new-test-queue").
 				Label(constants.WorkloadPriorityClassLabel, "new-test").
 				PodTemplateSpecQueue("new-test-queue").
diff --git a/pkg/controller/jobs/job/job_webhook_test.go b/pkg/controller/jobs/job/job_webhook_test.go
index 7c9b5eac3..a0bdd3f1b 100644
--- a/pkg/controller/jobs/job/job_webhook_test.go
+++ b/pkg/controller/jobs/job/job_webhook_test.go
@@ -27,6 +27,7 @@ import (
 	apivalidation "k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/utils/ptr"
 	jobsetapi "sigs.k8s.io/jobset/api/jobset/v1alpha2"
@@ -40,6 +41,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	testingutil "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
+	testingmpijob "sigs.k8s.io/kueue/pkg/util/testingjobs/mpijob"
 
 	// without this only the job framework is registered
 	_ "sigs.k8s.io/kueue/pkg/controller/jobs/mpijob"
@@ -549,6 +551,7 @@ func TestValidateUpdate(t *testing.T) {
 func TestDefault(t *testing.T) {
 	testcases := map[string]struct {
 		job                                    *batchv1.Job
+		objs                                   []runtime.Object
 		queues                                 []kueue.LocalQueue
 		clusterQueues                          []kueue.ClusterQueue
 		admissionCheck                         *kueue.AdmissionCheck
@@ -675,6 +678,9 @@ func TestDefault(t *testing.T) {
 			job: testingutil.MakeJob("test-job", metav1.NamespaceDefault).
 				OwnerReference("owner", kfmpi.SchemeGroupVersionKind).
 				Obj(),
+			objs: []runtime.Object{
+				testingmpijob.MakeMPIJob("owner", "default").UID("owner").Obj(),
+			},
 			want: testingutil.MakeJob("test-job", metav1.NamespaceDefault).
 				OwnerReference("owner", kfmpi.SchemeGroupVersionKind).
 				Obj(),
@@ -689,6 +695,7 @@ func TestDefault(t *testing.T) {
 				OwnerReference("owner", jobsetapi.SchemeGroupVersion.WithKind("JobSet")).
 				Queue("default").
 				Obj(),
+			wantErr: jobframework.ErrWorkloadOwnerNotFound,
 		},
 	}
 	for name, tc := range testcases {
@@ -699,10 +706,9 @@ func TestDefault(t *testing.T) {
 
 			ctx, _ := utiltesting.ContextWithLog(t)
 
-			clientBuilder := utiltesting.NewClientBuilder().
-				WithObjects(
-					utiltesting.MakeNamespace("default"),
-				)
+			clientBuilder := utiltesting.NewClientBuilder(kfmpi.AddToScheme).
+				WithObjects(utiltesting.MakeNamespace("default")).
+				WithRuntimeObjects(tc.objs...)
 			cl := clientBuilder.Build()
 			cqCache := cache.New(cl)
 			queueManager := queue.NewManager(cl, cqCache)
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
index 8f95a6409..64c7da36f 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler_test.go
@@ -109,6 +109,7 @@ func TestPodReconciler(t *testing.T) {
 				Label(leaderworkersetv1.SetNameLabelKey, "lws").
 				Label(leaderworkersetv1.GroupIndexLabelKey, "0").
 				Queue("queue").
+				ManagedByKueueLabel().
 				Group(GetWorkloadName(types.UID(testUID), "lws", "0")).
 				GroupTotalCount("1").
 				PrebuiltWorkload(GetWorkloadName(types.UID(testUID), "lws", "0")).
diff --git a/pkg/controller/jobs/pod/pod_webhook_test.go b/pkg/controller/jobs/pod/pod_webhook_test.go
index c1c2d7c0c..5ab93a9c5 100644
--- a/pkg/controller/jobs/pod/pod_webhook_test.go
+++ b/pkg/controller/jobs/pod/pod_webhook_test.go
@@ -21,11 +21,16 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	kfmpi "github.com/kubeflow/mpi-operator/pkg/apis/kubeflow/v2beta1"
+	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
@@ -33,12 +38,19 @@ import (
 	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
+	testingmpijob "sigs.k8s.io/kueue/pkg/util/testingjobs/mpijob"
+	"sigs.k8s.io/kueue/pkg/util/testingjobs/paddlejob"
 	testingpod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
+	testingpytorchjob "sigs.k8s.io/kueue/pkg/util/testingjobs/pytorchjob"
+	testingtfjob "sigs.k8s.io/kueue/pkg/util/testingjobs/tfjob"
+	testingxgboostjob "sigs.k8s.io/kueue/pkg/util/testingjobs/xgboostjob"
 
 	_ "sigs.k8s.io/kueue/pkg/controller/jobs/kubeflow/jobs"
 	_ "sigs.k8s.io/kueue/pkg/controller/jobs/mpijob"
@@ -56,19 +68,35 @@ func TestDefault(t *testing.T) {
 			},
 		},
 	}
+	defaultPodSelector := &metav1.LabelSelector{
+		MatchExpressions: []metav1.LabelSelectorRequirement{
+			{
+				Key:      corev1.LabelMetadataName,
+				Operator: metav1.LabelSelectorOpNotIn,
+				Values:   []string{"kube-system"},
+			},
+		},
+	}
+	defaultManagedJobsNamespaceSelector, err := metav1.LabelSelectorAsSelector(defaultNamespaceSelector)
+	if err != nil {
+		t.Fatalf("failed to parse namespace selector")
+	}
 
 	testCases := map[string]struct {
-		enableTopologyAwareScheduling bool
-
-		initObjects                []client.Object
-		pod                        *corev1.Pod
-		localQueueDefaulting       bool
-		defaultLqExist             bool
-		manageJobsWithoutQueueName bool
-		namespaceSelector          *metav1.LabelSelector
-		podSelector                *metav1.LabelSelector
-		enableIntegrations         []string
-		want                       *corev1.Pod
+		enableTopologyAwareScheduling      bool
+		enableManagedJobsNamespaceSelector bool
+
+		initObjects                  []client.Object
+		pod                          *corev1.Pod
+		localQueueDefaulting         bool
+		defaultLqExist               bool
+		manageJobsWithoutQueueName   bool
+		managedJobsNamespaceSelector labels.Selector
+		namespaceSelector            *metav1.LabelSelector
+		podSelector                  *metav1.LabelSelector
+		enableIntegrations           []string
+		want                         *corev1.Pod
+		wantErr                      error
 	}{
 		"pod with queue nil ns selector": {
 			initObjects: []client.Object{defaultNamespace},
@@ -110,7 +138,10 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (Job) while not enabled": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				testingjob.MakeJob("parent-job", defaultNamespace.Name).UID("parent-job").Queue("test-queue").Obj(),
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -126,7 +157,10 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (Job)": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				testingjob.MakeJob("parent-job", defaultNamespace.Name).UID("parent-job").Queue("test-queue").Obj(),
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -140,7 +174,19 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (RayCluster)": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				&rayv1.RayCluster{
+					ObjectMeta: metav1.ObjectMeta{
+						UID:       types.UID("parent-ray-cluster"),
+						Name:      "parent-ray-cluster",
+						Namespace: defaultNamespace.Name,
+						Labels: map[string]string{
+							constants.QueueLabel: "test-queue",
+						},
+					},
+				},
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -154,7 +200,10 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (MPIJob)": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				testingmpijob.MakeMPIJob("parent-mpi-job", defaultNamespace.Name).UID("parent-mpi-job").Queue("test-queue").Obj(),
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -174,7 +223,10 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (PyTorchJob)": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				testingpytorchjob.MakePyTorchJob("parent-pytorch-job", defaultNamespace.Name).UID("parent-pytorch-job").Queue("test-queue").Obj(),
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -194,7 +246,10 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (TFJob)": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				testingtfjob.MakeTFJob("parent-tf-job", defaultNamespace.Name).UID("parent-tf-job").Queue("test-queue").Obj(),
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -214,7 +269,10 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (XGBoostJob)": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				testingxgboostjob.MakeXGBoostJob("parent-xgboost-job", defaultNamespace.Name).UID("parent-xgboost-job").Queue("test-queue").Obj(),
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -234,7 +292,10 @@ func TestDefault(t *testing.T) {
 				Obj(),
 		},
 		"pod with owner managed by kueue (PaddleJob)": {
-			initObjects:       []client.Object{defaultNamespace},
+			initObjects: []client.Object{
+				defaultNamespace,
+				paddlejob.MakePaddleJob("parent-paddle-job", defaultNamespace.Name).UID("parent-paddle-job").Queue("test-queue").Obj(),
+			},
 			podSelector:       &metav1.LabelSelector{},
 			namespaceSelector: defaultNamespaceSelector,
 			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
@@ -305,6 +366,29 @@ func TestDefault(t *testing.T) {
 				TopologySchedulingGate().
 				Obj(),
 		},
+		"pod with TAS and PodGroupPodIndexLabelAnnotation": {
+			enableTopologyAwareScheduling: true,
+			initObjects:                   []client.Object{defaultNamespace},
+			podSelector:                   &metav1.LabelSelector{},
+			namespaceSelector:             defaultNamespaceSelector,
+			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Queue("test-queue").
+				Label("test-label", "test-value").
+				Annotation(kueuealpha.PodGroupPodIndexLabelAnnotation, "test-label").
+				Annotation(kueuealpha.PodSetRequiredTopologyAnnotation, "block").
+				Obj(),
+			want: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Queue("test-queue").
+				Annotation(kueuealpha.PodGroupPodIndexLabelAnnotation, "test-label").
+				Annotation(kueuealpha.PodSetRequiredTopologyAnnotation, "block").
+				Label("test-label", "test-value").
+				Label(kueuealpha.PodGroupPodIndexLabel, "test-value").
+				ManagedByKueueLabel().
+				KueueFinalizer().
+				KueueSchedulingGate().
+				TopologySchedulingGate().
+				Obj(),
+		},
 		"LocalQueueDefaulting enabled, default queue is created, pod has no queue label": {
 			initObjects:          []client.Object{defaultNamespace},
 			localQueueDefaulting: true,
@@ -347,55 +431,126 @@ func TestDefault(t *testing.T) {
 				KueueFinalizer().
 				Obj(),
 		},
+		"ManagedJobsNamespaceSelector is enabled and the namespace matches the selector": {
+			enableManagedJobsNamespaceSelector: true,
+			initObjects:                        []client.Object{defaultNamespace},
+			managedJobsNamespaceSelector:       defaultManagedJobsNamespaceSelector,
+			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Queue("queue").
+				Obj(),
+			want: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Queue("queue").
+				ManagedByKueueLabel().
+				KueueSchedulingGate().
+				KueueFinalizer().
+				Obj(),
+		},
+		"ManagedJobsNamespaceSelector is enabled but doesn’t match the managedJobsNamespaceSelector": {
+			enableManagedJobsNamespaceSelector: true,
+			initObjects: []client.Object{
+				utiltesting.MakeNamespaceWrapper("kube-system").Label(corev1.LabelMetadataName, "kube-system").Obj(),
+			},
+			managedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
+			pod: testingpod.MakePod("test-pod", "kube-system").
+				Queue("queue").
+				Obj(),
+			want: testingpod.MakePod("test-pod", "kube-system").
+				Queue("queue").
+				Obj(),
+		},
+		"the namespace matches the namespace selector": {
+			initObjects:       []client.Object{defaultNamespace},
+			namespaceSelector: defaultNamespaceSelector,
+			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Queue("queue").
+				Obj(),
+			want: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Queue("queue").
+				ManagedByKueueLabel().
+				KueueSchedulingGate().
+				KueueFinalizer().
+				Obj(),
+		},
+		"the namespace doesn't match the namespace selector": {
+			initObjects: []client.Object{
+				utiltesting.MakeNamespaceWrapper("kube-system").Label(corev1.LabelMetadataName, "kube-system").Obj(),
+			},
+			namespaceSelector: defaultNamespaceSelector,
+			pod: testingpod.MakePod("test-pod", "kube-system").
+				Queue("queue").
+				Obj(),
+			want: testingpod.MakePod("test-pod", "kube-system").
+				Queue("queue").
+				Obj(),
+		},
+		"the pod matches the pod selector": {
+			initObjects: []client.Object{defaultNamespace},
+			podSelector: defaultPodSelector,
+			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Label(corev1.LabelMetadataName, "test-pod").
+				Queue("queue").
+				Obj(),
+			want: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Label(corev1.LabelMetadataName, "test-pod").
+				Queue("queue").
+				ManagedByKueueLabel().
+				KueueSchedulingGate().
+				KueueFinalizer().
+				Obj(),
+		},
+		"the pod doesn't match the pod selector": {
+			initObjects: []client.Object{defaultNamespace},
+			podSelector: defaultPodSelector,
+			pod: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Label(corev1.LabelMetadataName, "kube-system").
+				Queue("queue").
+				Obj(),
+			want: testingpod.MakePod("test-pod", defaultNamespace.Name).
+				Label(corev1.LabelMetadataName, "kube-system").
+				Queue("queue").
+				Obj(),
+		},
 	}
 
-	for _, managedJobsFeatureGate := range []bool{false, true} {
-		for name, tc := range testCases {
-			if managedJobsFeatureGate {
-				name += " managedJobsNamespaceSelector"
-			}
-			t.Run(name, func(t *testing.T) {
-				features.SetFeatureGateDuringTest(t, features.TopologyAwareScheduling, tc.enableTopologyAwareScheduling)
-				features.SetFeatureGateDuringTest(t, features.ManagedJobsNamespaceSelector, managedJobsFeatureGate)
-				features.SetFeatureGateDuringTest(t, features.LocalQueueDefaulting, tc.localQueueDefaulting)
-				t.Cleanup(jobframework.EnableIntegrationsForTest(t, tc.enableIntegrations...))
-				builder := utiltesting.NewClientBuilder()
-				builder = builder.WithObjects(tc.initObjects...)
-				cli := builder.Build()
-				mjls, err := metav1.LabelSelectorAsSelector(tc.namespaceSelector)
-				if err != nil {
-					t.Errorf("failed to parse namespace selector")
-				}
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			features.SetFeatureGateDuringTest(t, features.TopologyAwareScheduling, tc.enableTopologyAwareScheduling)
+			features.SetFeatureGateDuringTest(t, features.ManagedJobsNamespaceSelector, tc.enableManagedJobsNamespaceSelector)
+			features.SetFeatureGateDuringTest(t, features.LocalQueueDefaulting, tc.localQueueDefaulting)
+			t.Cleanup(jobframework.EnableIntegrationsForTest(t, tc.enableIntegrations...))
+			builder := utiltesting.NewClientBuilder(rayv1.AddToScheme, kfmpi.AddToScheme, kftraining.AddToScheme, appsv1.AddToScheme)
+			builder = builder.WithObjects(tc.initObjects...)
+			cli := builder.Build()
 
-				cqCache := cache.New(cli)
-				queueManager := queue.NewManager(cli, cqCache)
+			cqCache := cache.New(cli)
+			queueManager := queue.NewManager(cli, cqCache)
 
-				ctx, _ := utiltesting.ContextWithLog(t)
+			ctx, _ := utiltesting.ContextWithLog(t)
 
-				if tc.defaultLqExist {
-					if err := queueManager.AddLocalQueue(ctx, utiltesting.MakeLocalQueue("default", defaultNamespace.Name).
-						ClusterQueue("cluster-queue").Obj()); err != nil {
-						t.Fatalf("failed to create default local queue: %s", err)
-					}
+			if tc.defaultLqExist {
+				if err := queueManager.AddLocalQueue(ctx, utiltesting.MakeLocalQueue("default", defaultNamespace.Name).
+					ClusterQueue("cluster-queue").Obj()); err != nil {
+					t.Fatalf("failed to create default local queue: %s", err)
 				}
+			}
 
-				w := &PodWebhook{
-					client:                       cli,
-					queues:                       queueManager,
-					manageJobsWithoutQueueName:   tc.manageJobsWithoutQueueName,
-					managedJobsNamespaceSelector: mjls,
-					namespaceSelector:            tc.namespaceSelector,
-					podSelector:                  tc.podSelector,
-				}
+			w := &PodWebhook{
+				client:                       cli,
+				queues:                       queueManager,
+				manageJobsWithoutQueueName:   tc.manageJobsWithoutQueueName,
+				managedJobsNamespaceSelector: tc.managedJobsNamespaceSelector,
+				namespaceSelector:            tc.namespaceSelector,
+				podSelector:                  tc.podSelector,
+			}
 
-				if err := w.Default(ctx, tc.pod); err != nil {
-					t.Errorf("failed to set defaults for v1/pod: %s", err)
-				}
-				if diff := cmp.Diff(tc.want, tc.pod); len(diff) != 0 {
-					t.Errorf("Default() mismatch (-want,+got):\n%s", diff)
-				}
-			})
-		}
+			if err := w.Default(ctx, tc.pod); err != nil {
+				t.Errorf("failed to set defaults for v1/pod: %s", err)
+			}
+
+			if diff := cmp.Diff(tc.want, tc.pod); len(diff) != 0 {
+				t.Errorf("Pod mismatch (-want,+got):\n%s", diff)
+			}
+		})
 	}
 }
 
diff --git a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
index 55930510b..cedef8e8e 100644
--- a/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
+++ b/pkg/controller/jobs/statefulset/statefulset_webhook_test.go
@@ -26,6 +26,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
@@ -40,6 +41,8 @@ import (
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	testingappwrapper "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
+	testingleaderworkerset "sigs.k8s.io/kueue/pkg/util/testingjobs/leaderworkerset"
 	testingstatefulset "sigs.k8s.io/kueue/pkg/util/testingjobs/statefulset"
 )
 
@@ -62,6 +65,7 @@ func TestDefault(t *testing.T) {
 				Replicas(10).
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
+				PodTemplateManagedByKueue().
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(10).
@@ -82,6 +86,7 @@ func TestDefault(t *testing.T) {
 				Queue("test-queue").
 				Label(constants.WorkloadPriorityClassLabel, "test").
 				PodTemplateSpecQueue("test-queue").
+				PodTemplateManagedByKueue().
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecLabel(constants.WorkloadPriorityClassLabel, "test").
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
@@ -98,6 +103,7 @@ func TestDefault(t *testing.T) {
 				Obj(),
 			want: testingstatefulset.MakeStatefulSet("test-pod", "").
 				Queue("test-queue").
+				PodTemplateManagedByKueue().
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(1).
 				PodTemplateSpecQueue("test-queue").
@@ -114,6 +120,7 @@ func TestDefault(t *testing.T) {
 			want: testingstatefulset.MakeStatefulSet("test-pod", "default").
 				Queue("default").
 				PodTemplateSpecQueue("default").
+				PodTemplateManagedByKueue().
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(1).
@@ -129,6 +136,7 @@ func TestDefault(t *testing.T) {
 			want: testingstatefulset.MakeStatefulSet("test-pod", "").
 				Queue("test-queue").
 				PodTemplateSpecQueue("test-queue").
+				PodTemplateManagedByKueue().
 				PodTemplateAnnotation(podconstants.SuspendedByParentAnnotation, FrameworkName).
 				PodTemplateSpecPodGroupNameLabel("test-pod", "", gvk).
 				PodTemplateSpecPodGroupTotalCountAnnotation(1).
@@ -226,6 +234,7 @@ func TestValidateCreate(t *testing.T) {
 func TestValidateUpdate(t *testing.T) {
 	testCases := map[string]struct {
 		integrations []string
+		objs         []runtime.Object
 		oldObj       *appsv1.StatefulSet
 		newObj       *appsv1.StatefulSet
 		wantErr      error
@@ -431,17 +440,19 @@ func TestValidateUpdate(t *testing.T) {
 		},
 		"change in replicas (scale up with AppWrapper ownerReference while the previous scaling operation is still in progress)": {
 			integrations: []string{appwrapper.FrameworkName},
+			objs: []runtime.Object{
+				testingappwrapper.MakeAppWrapper("test-app-wrapper", "test-ns").
+					UID("test-app-wrapper").
+					Queue("test-queue").
+					Obj(),
+			},
 			oldObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
 				Queue("test-queue").
 				Replicas(0).
 				StatusReplicas(3).
 				Obj(),
 			newObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
-				WithOwnerReference(metav1.OwnerReference{
-					APIVersion: awv1beta2.GroupVersion.String(),
-					Kind:       awv1beta2.AppWrapperKind,
-					Controller: ptr.To(true),
-				}).
+				OwnerReference("test-app-wrapper", awv1beta2.GroupVersion.WithKind(awv1beta2.AppWrapperKind)).
 				Queue("test-queue").
 				Replicas(3).
 				StatusReplicas(1).
@@ -449,33 +460,37 @@ func TestValidateUpdate(t *testing.T) {
 		},
 		"change in replicas (scale up with AppWrapper ownerReference)": {
 			integrations: []string{appwrapper.FrameworkName},
+			objs: []runtime.Object{
+				testingappwrapper.MakeAppWrapper("test-app-wrapper", "test-ns").
+					UID("test-app-wrapper").
+					Queue("test-queue").
+					Obj(),
+			},
 			oldObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
 				Queue("test-queue").
 				Replicas(3).
 				Obj(),
 			newObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
-				WithOwnerReference(metav1.OwnerReference{
-					APIVersion: awv1beta2.GroupVersion.String(),
-					Kind:       awv1beta2.AppWrapperKind,
-					Controller: ptr.To(true),
-				}).
+				OwnerReference("test-app-wrapper", awv1beta2.GroupVersion.WithKind(awv1beta2.AppWrapperKind)).
 				Queue("test-queue").
 				Replicas(4).
 				Obj(),
 		},
 		"change in replicas (scale up with LeaderWorkerSet ownerReference while the previous scaling operation is still in progress)": {
 			integrations: []string{leaderworkerset.FrameworkName},
+			objs: []runtime.Object{
+				testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "test-ns").
+					UID("test-lws").
+					Queue("test-queue").
+					Obj(),
+			},
 			oldObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
 				Queue("test-queue").
 				Replicas(0).
 				StatusReplicas(3).
 				Obj(),
 			newObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
-				WithOwnerReference(metav1.OwnerReference{
-					APIVersion: leaderworkersetv1.GroupVersion.String(),
-					Kind:       "LeaderWorkerSet",
-					Controller: ptr.To(true),
-				}).
+				OwnerReference("test-lws", leaderworkersetv1.GroupVersion.WithKind("LeaderWorkerSet")).
 				Queue("test-queue").
 				Replicas(3).
 				StatusReplicas(1).
@@ -483,16 +498,18 @@ func TestValidateUpdate(t *testing.T) {
 		},
 		"change in replicas (scale up with LeaderWorkerSet ownerReference)": {
 			integrations: []string{leaderworkerset.FrameworkName},
+			objs: []runtime.Object{
+				testingleaderworkerset.MakeLeaderWorkerSet("test-lws", "test-ns").
+					UID("test-lws").
+					Queue("test-queue").
+					Obj(),
+			},
 			oldObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
 				Queue("test-queue").
 				Replicas(3).
 				Obj(),
 			newObj: testingstatefulset.MakeStatefulSet("test-sts", "test-ns").
-				WithOwnerReference(metav1.OwnerReference{
-					APIVersion: leaderworkersetv1.GroupVersion.String(),
-					Kind:       "LeaderWorkerSet",
-					Controller: ptr.To(true),
-				}).
+				OwnerReference("test-lws", leaderworkersetv1.GroupVersion.WithKind("LeaderWorkerSet")).
 				Queue("test-queue").
 				Replicas(4).
 				Obj(),
@@ -612,7 +629,13 @@ func TestValidateUpdate(t *testing.T) {
 			t.Cleanup(jobframework.EnableIntegrationsForTest(t, tc.integrations...))
 			ctx := t.Context()
 
-			wh := &Webhook{}
+			client := utiltesting.NewClientBuilder(awv1beta2.AddToScheme, leaderworkersetv1.AddToScheme).
+				WithRuntimeObjects(tc.objs...).
+				Build()
+
+			wh := &Webhook{
+				client: client,
+			}
 
 			_, err := wh.ValidateUpdate(ctx, tc.oldObj, tc.newObj)
 			if diff := cmp.Diff(tc.wantErr, err, cmpopts.IgnoreFields(field.Error{}, "BadValue", "Detail")); diff != "" {
diff --git a/pkg/util/testing/wrappers.go b/pkg/util/testing/wrappers.go
index 973ab5d54..bf4a8cc18 100644
--- a/pkg/util/testing/wrappers.go
+++ b/pkg/util/testing/wrappers.go
@@ -1546,6 +1546,10 @@ func MakeNamespaceWrapper(name string) *NamespaceWrapper {
 	}
 }
 
+func (w *NamespaceWrapper) Clone() *NamespaceWrapper {
+	return &NamespaceWrapper{Namespace: *w.DeepCopy()}
+}
+
 func (w *NamespaceWrapper) Obj() *corev1.Namespace {
 	return &w.Namespace
 }
diff --git a/pkg/util/testingjobs/deployment/wrappers.go b/pkg/util/testingjobs/deployment/wrappers.go
index 3cebbe778..a932bb47d 100644
--- a/pkg/util/testingjobs/deployment/wrappers.go
+++ b/pkg/util/testingjobs/deployment/wrappers.go
@@ -23,8 +23,10 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 
-	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/constants"
+	controllerconstants "sigs.k8s.io/kueue/pkg/controller/constants"
 )
 
 // DeploymentWrapper wraps a Deployment.
@@ -82,7 +84,7 @@ func (d *DeploymentWrapper) Label(k, v string) *DeploymentWrapper {
 
 // Queue updates the queue name of the Deployment
 func (d *DeploymentWrapper) Queue(q string) *DeploymentWrapper {
-	return d.Label(constants.QueueLabel, q)
+	return d.Label(controllerconstants.QueueLabel, q)
 }
 
 // Name updated the name of the Deployment
@@ -91,6 +93,12 @@ func (d *DeploymentWrapper) Name(n string) *DeploymentWrapper {
 	return d
 }
 
+// UID updates the uid of the Deployment.
+func (d *DeploymentWrapper) UID(uid string) *DeploymentWrapper {
+	d.ObjectMeta.UID = types.UID(uid)
+	return d
+}
+
 // Image sets an image to the default container.
 func (d *DeploymentWrapper) Image(image string, args []string) *DeploymentWrapper {
 	d.Spec.Template.Spec.Containers[0].Image = image
@@ -153,7 +161,11 @@ func (d *DeploymentWrapper) PodTemplateAnnotation(k, v string) *DeploymentWrappe
 
 // PodTemplateSpecQueue updates the queue name of the pod template spec of the Deployment
 func (d *DeploymentWrapper) PodTemplateSpecQueue(q string) *DeploymentWrapper {
-	return d.PodTemplateSpecLabel(constants.QueueLabel, q)
+	return d.PodTemplateSpecLabel(controllerconstants.QueueLabel, q)
+}
+
+func (d *DeploymentWrapper) PodTemplateSpecManagedByKueue() *DeploymentWrapper {
+	return d.PodTemplateSpecLabel(constants.ManagedByKueueLabelKey, constants.ManagedByKueueLabelValue)
 }
 
 func (d *DeploymentWrapper) TerminationGracePeriod(seconds int64) *DeploymentWrapper {
diff --git a/pkg/util/testingjobs/statefulset/wrappers.go b/pkg/util/testingjobs/statefulset/wrappers.go
index bb8b8dafa..b3b16bed3 100644
--- a/pkg/util/testingjobs/statefulset/wrappers.go
+++ b/pkg/util/testingjobs/statefulset/wrappers.go
@@ -25,11 +25,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
 
 	"sigs.k8s.io/kueue/apis/kueue/v1alpha1"
-	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/constants"
+	controllerconstants "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
+	"sigs.k8s.io/kueue/pkg/util/testing"
 )
 
 // StatefulSetWrapper wraps a StatefulSet.
@@ -87,7 +90,7 @@ func (ss *StatefulSetWrapper) Label(k, v string) *StatefulSetWrapper {
 
 // Queue updates the queue name of the StatefulSet
 func (ss *StatefulSetWrapper) Queue(q string) *StatefulSetWrapper {
-	return ss.Label(constants.QueueLabel, q)
+	return ss.Label(controllerconstants.QueueLabel, q)
 }
 
 // Name updated the name of the StatefulSet
@@ -96,8 +99,15 @@ func (ss *StatefulSetWrapper) Name(n string) *StatefulSetWrapper {
 	return ss
 }
 
-func (ss *StatefulSetWrapper) WithOwnerReference(ownerReference metav1.OwnerReference) *StatefulSetWrapper {
-	ss.OwnerReferences = append(ss.OwnerReferences, ownerReference)
+// UID updates the uid of the StatefulSet
+func (ss *StatefulSetWrapper) UID(uid string) *StatefulSetWrapper {
+	ss.ObjectMeta.UID = types.UID(uid)
+	return ss
+}
+
+// OwnerReference adds a ownerReference to the StatefulSet.
+func (ss *StatefulSetWrapper) OwnerReference(ownerName string, ownerGVK schema.GroupVersionKind) *StatefulSetWrapper {
+	testing.AppendOwnerReference(&ss.StatefulSet, ownerGVK, ownerName, ownerName, ptr.To(true), ptr.To(true))
 	return ss
 }
 
@@ -136,7 +146,11 @@ func (ss *StatefulSetWrapper) PodTemplateSpecAnnotation(k, v string) *StatefulSe
 
 // PodTemplateSpecQueue updates the queue name of the pod template spec of the StatefulSet
 func (ss *StatefulSetWrapper) PodTemplateSpecQueue(q string) *StatefulSetWrapper {
-	return ss.PodTemplateSpecLabel(constants.QueueLabel, q)
+	return ss.PodTemplateSpecLabel(controllerconstants.QueueLabel, q)
+}
+
+func (ss *StatefulSetWrapper) PodTemplateManagedByKueue() *StatefulSetWrapper {
+	return ss.PodTemplateSpecLabel(constants.ManagedByKueueLabelKey, constants.ManagedByKueueLabelValue)
 }
 
 func (ss *StatefulSetWrapper) Replicas(r int32) *StatefulSetWrapper {
@@ -224,5 +238,5 @@ func (ss *StatefulSetWrapper) TerminationGracePeriod(seconds int64) *StatefulSet
 
 // WorkloadPriorityClass sets workloadpriorityclass.
 func (ss *StatefulSetWrapper) WorkloadPriorityClass(wpc string) *StatefulSetWrapper {
-	return ss.Label(constants.WorkloadPriorityClassLabel, wpc)
+	return ss.Label(controllerconstants.WorkloadPriorityClassLabel, wpc)
 }
diff --git a/test/e2e/customconfigs/managejobswithoutqueuename_test.go b/test/e2e/customconfigs/managejobswithoutqueuename_test.go
index f9967833c..1805e7d6b 100644
--- a/test/e2e/customconfigs/managejobswithoutqueuename_test.go
+++ b/test/e2e/customconfigs/managejobswithoutqueuename_test.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/utils/strings/slices"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/jobset/api/jobset/v1alpha2"
+	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
 
 	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
@@ -36,12 +37,15 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	"sigs.k8s.io/kueue/pkg/controller/jobs/jobset"
+	"sigs.k8s.io/kueue/pkg/controller/jobs/leaderworkerset"
 	podcontroller "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
+	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	awtesting "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
 	testingdeploy "sigs.k8s.io/kueue/pkg/util/testingjobs/deployment"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingjobset "sigs.k8s.io/kueue/pkg/util/testingjobs/jobset"
+	leaderworkersettesting "sigs.k8s.io/kueue/pkg/util/testingjobs/leaderworkerset"
 	testingpod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
 	testingsts "sigs.k8s.io/kueue/pkg/util/testingjobs/statefulset"
 	"sigs.k8s.io/kueue/test/util"
@@ -625,6 +629,97 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
 			})
 		})
+
+		ginkgo.It("should not suspend the pods created by a LeaderWorkerSet in the test namespace with queue-name label", func() {
+			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
+				Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+				Size(3).
+				Replicas(1).
+				RequestAndLimit(corev1.ResourceCPU, "200m").
+				TerminationGracePeriod(1).
+				Queue(localQueue.Name).
+				Obj()
+			ginkgo.By("Create a LeaderWorkerSet", func() {
+				util.MustCreate(ctx, k8sClient, lws)
+			})
+
+			ginkgo.By("check that only one workload is created and admitted", func() {
+				createdWorkloads := &kueue.WorkloadList{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, createdWorkloads, client.InNamespace(lws.Namespace))).To(gomega.Succeed())
+					g.Expect(createdWorkloads.Items).To(gomega.HaveLen(1))
+					g.Expect(createdWorkloads.Items[0].Name).To(gomega.Equal(leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0")))
+					g.Expect(createdWorkloads.Items[0].Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadAdmitted))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("waiting for replicas is ready", func() {
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(testing.HaveConditionStatusTrue("Available"))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying that pods are running", func() {
+				pods := &corev1.PodList{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+						leaderworkersetv1.SetNameLabelKey: lws.Name,
+					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+					g.Expect(pods.Items).To(gomega.HaveLen(3))
+					for _, pod := range pods.Items {
+						g.Expect(pod.Status.Phase).Should(gomega.Equal(corev1.PodRunning))
+					}
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+
+		ginkgo.It("should suspend the pods created by a LeaderWorkerSet in the test namespace without queue-name label", func() {
+			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
+				Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+				Size(3).
+				Replicas(1).
+				RequestAndLimit(corev1.ResourceCPU, "200m").
+				TerminationGracePeriod(1).
+				Obj()
+			ginkgo.By("Create a LeaderWorkerSet", func() {
+				util.MustCreate(ctx, k8sClient, lws)
+			})
+
+			ginkgo.By("check that only one workload is created and is not admitted", func() {
+				createdWorkloads := &kueue.WorkloadList{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, createdWorkloads, client.InNamespace(lws.Namespace))).To(gomega.Succeed())
+					g.Expect(createdWorkloads.Items).To(gomega.HaveLen(1))
+					g.Expect(createdWorkloads.Items[0].Name).To(gomega.Equal(leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0")))
+					g.Expect(createdWorkloads.Items[0].Status.Conditions).To(testing.HaveConditionStatusFalse(kueue.WorkloadQuotaReserved))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verify that replicas is not ready", func() {
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(0)))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying that pods are running and gated", func() {
+				pods := &corev1.PodList{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+						leaderworkersetv1.SetNameLabelKey: lws.Name,
+					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+					g.Expect(pods.Items).To(gomega.HaveLen(3))
+					for _, pod := range pods.Items {
+						g.Expect(pod.Status.Phase).Should(gomega.Equal(corev1.PodPending))
+						g.Expect(utilpod.HasGate(&pod, podcontroller.SchedulingGateName)).Should(gomega.BeTrue())
+					}
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
 	})
 })
 
diff --git a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
index bd59b7724..e47f7bb78 100644
--- a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
@@ -24,7 +24,6 @@ import (
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
-	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -43,6 +42,7 @@ import (
 	podconstants "sigs.k8s.io/kueue/pkg/controller/jobs/pod/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/testing"
+	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingnode "sigs.k8s.io/kueue/pkg/util/testingjobs/node"
 	testingpod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
 	"sigs.k8s.io/kueue/pkg/workload"
@@ -349,15 +349,12 @@ var _ = ginkgo.Describe("Pod controller", ginkgo.Ordered, ginkgo.ContinueOnFailu
 			})
 
 			ginkgo.When("Pod owner is managed by Kueue", func() {
-				var pod *corev1.Pod
-				ginkgo.BeforeEach(func() {
-					pod = testingpod.MakePod(podName, ns.Name).
-						Queue("test-queue").
-						OwnerReference("parent-job", batchv1.SchemeGroupVersion.WithKind("Job")).
-						Obj()
-				})
-
 				ginkgo.It("Should skip the pod", func() {
+					job := testingjob.MakeJob("parent-job", ns.Name).Queue(lq.Name).Obj()
+					util.MustCreate(ctx, k8sClient, job)
+
+					pod := testingpod.MakePod(podName, ns.Name).Queue(lq.Name).Obj()
+					gomega.Expect(controllerutil.SetControllerReference(job, pod, k8sClient.Scheme())).To(gomega.Succeed())
 					util.MustCreate(ctx, k8sClient, pod)
 
 					createdPod := &corev1.Pod{}
diff --git a/test/integration/singlecluster/controller/jobs/pod/suite_test.go b/test/integration/singlecluster/controller/jobs/pod/suite_test.go
index 6a3112ca2..1adf77e55 100644
--- a/test/integration/singlecluster/controller/jobs/pod/suite_test.go
+++ b/test/integration/singlecluster/controller/jobs/pod/suite_test.go
@@ -123,6 +123,8 @@ func managerSetup(
 		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
 
+		err = job.SetupWebhook(mgr, opts...)
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 		err = pod.SetupWebhook(mgr, opts...)
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 		failedWebhook, err := webhooks.Setup(mgr)