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/pkg/controller/jobframework/integrationmanager.go b/pkg/controller/jobframework/integrationmanager.go
index ad9d60459..812404a07 100644
--- a/pkg/controller/jobframework/integrationmanager.go
+++ b/pkg/controller/jobframework/integrationmanager.go
@@ -74,10 +74,6 @@ type IntegrationCallbacks struct {
 	// AddToScheme adds any additional types to the controllers manager's scheme
 	// (this callback is optional)
 	AddToScheme func(s *runtime.Scheme) error
-	// Returns true if the provided owner reference identifies an object
-	// managed by this integration
-	// (this callback is optional)
-	IsManagingObjectsOwner func(ref *metav1.OwnerReference) bool
 	// CanSupportIntegration returns true if the integration meets any additional condition
 	// like the Kubernetes version.
 	CanSupportIntegration func(opts ...Option) (bool, error)
@@ -87,6 +83,21 @@ type IntegrationCallbacks struct {
 	DependencyList []string
 }
 
+func (i *IntegrationCallbacks) getGVK() schema.GroupVersionKind {
+	if i.NewJob != nil {
+		return i.NewJob().GVK()
+	}
+	return i.GVK
+}
+
+func (i *IntegrationCallbacks) matchingGVK(gvk schema.GroupVersionKind) bool {
+	return i.getGVK() == gvk
+}
+
+func (i *IntegrationCallbacks) matchingOwnerReference(ownerRef *metav1.OwnerReference) bool {
+	return ownerReferenceMatchingGVK(ownerRef, i.getGVK())
+}
+
 type integrationManager struct {
 	names                []string
 	integrations         map[string]IntegrationCallbacks
@@ -190,13 +201,12 @@ func (m *integrationManager) getList() []string {
 func (m *integrationManager) getJobTypeForOwner(ownerRef *metav1.OwnerReference) runtime.Object {
 	for jobKey := range m.getEnabledIntegrations() {
 		cbs, found := m.integrations[jobKey]
-		if found && cbs.IsManagingObjectsOwner != nil && cbs.IsManagingObjectsOwner(ownerRef) {
+		if found && cbs.matchingOwnerReference(ownerRef) {
 			return cbs.JobType
 		}
 	}
 	for _, jt := range m.externalIntegrations {
-		apiVersion, kind := jt.GetObjectKind().GroupVersionKind().ToAPIVersionAndKind()
-		if ownerRef.Kind == kind && ownerRef.APIVersion == apiVersion {
+		if ownerReferenceMatchingGVK(ownerRef, jt.GetObjectKind().GroupVersionKind()) {
 			return jt
 		}
 	}
@@ -222,20 +232,15 @@ func (m *integrationManager) checkEnabledListDependencies(enabledSet sets.Set[st
 }
 
 // isOwnerIntegrationEnabled returns true if the provided owner is managed by an enabled integration.
-func (m *integrationManager) isOwnerIntegrationEnabled(owner *metav1.OwnerReference) bool {
-	ownerGV, err := schema.ParseGroupVersion(owner.APIVersion)
-	if err != nil {
-		return false
-	}
-	gvk := ownerGV.WithKind(owner.Kind)
+func (m *integrationManager) isOwnerIntegrationEnabled(ownerRef *metav1.OwnerReference) bool {
 	for jobKey := range m.getEnabledIntegrations() {
 		cbs, found := m.integrations[jobKey]
-		if found && matchingGVK(cbs, gvk) {
+		if found && cbs.matchingOwnerReference(ownerRef) {
 			return true
 		}
 	}
 	for _, jt := range m.externalIntegrations {
-		if jt.GetObjectKind().GroupVersionKind() == gvk {
+		if ownerReferenceMatchingGVK(ownerRef, jt.GetObjectKind().GroupVersionKind()) {
 			return true
 		}
 	}
@@ -292,19 +297,16 @@ func GetIntegration(name string) (IntegrationCallbacks, bool) {
 func GetIntegrationByGVK(gvk schema.GroupVersionKind) (IntegrationCallbacks, bool) {
 	for _, name := range manager.getList() {
 		integration, ok := GetIntegration(name)
-		if ok && matchingGVK(integration, gvk) {
+		if ok && integration.matchingGVK(gvk) {
 			return integration, true
 		}
 	}
 	return IntegrationCallbacks{}, false
 }
 
-func matchingGVK(integration IntegrationCallbacks, gvk schema.GroupVersionKind) bool {
-	if integration.NewJob != nil {
-		return gvk == integration.NewJob().GVK()
-	} else {
-		return gvk == integration.GVK
-	}
+func ownerReferenceMatchingGVK(ownerRef *metav1.OwnerReference, gvk schema.GroupVersionKind) bool {
+	apiVersion, kind := gvk.ToAPIVersionAndKind()
+	return ownerRef.APIVersion == apiVersion && ownerRef.Kind == kind
 }
 
 // GetIntegrationsList returns the list of currently registered frameworks.
@@ -327,15 +329,6 @@ func IsOwnerManagedByKueueForObject(obj client.Object) bool {
 	return false
 }
 
-// IsOwnerIntegrationEnabled returns true if the provided owner is managed by an enabled integration.
-func IsOwnerIntegrationEnabled(owner *metav1.OwnerReference) bool {
-	// This function should be redundant with IsOwnerManagedByKueue, but currently is not.
-	// The difference is caused because the Deployment and StatefulState integrations do not register
-	// an IsManagingObjectsOwner function.  We should attempt to register these integrations properly,
-	// adjust GenericJobReconciler to handle this, and go back to only one function to answer this question.
-	return manager.isOwnerIntegrationEnabled(owner)
-}
-
 // GetEmptyOwnerObject returns an empty object of the owner's type,
 // returns nil if the owner is not manageable by kueue.
 func GetEmptyOwnerObject(owner *metav1.OwnerReference) client.Object {
diff --git a/pkg/controller/jobs/appwrapper/appwrapper_controller.go b/pkg/controller/jobs/appwrapper/appwrapper_controller.go
index d0db252a1..6869fbcbd 100644
--- a/pkg/controller/jobs/appwrapper/appwrapper_controller.go
+++ b/pkg/controller/jobs/appwrapper/appwrapper_controller.go
@@ -20,12 +20,10 @@ import (
 	"context"
 	"fmt"
 	"strconv"
-	"strings"
 
 	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	awutils "github.com/project-codeflare/appwrapper/pkg/utils"
 	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -57,15 +55,14 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		NewJob:                 NewJob,
-		GVK:                    gvk,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupAppWrapperWebhook,
-		JobType:                &awv1beta2.AppWrapper{},
-		SetupIndexes:           SetupIndexes,
-		AddToScheme:            awv1beta2.AddToScheme,
-		IsManagingObjectsOwner: isAppWrapper,
-		MultiKueueAdapter:      &multiKueueAdapter{},
+		NewJob:            NewJob,
+		GVK:               gvk,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupAppWrapperWebhook,
+		JobType:           &awv1beta2.AppWrapper{},
+		SetupIndexes:      SetupIndexes,
+		AddToScheme:       awv1beta2.AddToScheme,
+		MultiKueueAdapter: &multiKueueAdapter{},
 	}))
 }
 
@@ -85,10 +82,6 @@ func NewJob() jobframework.GenericJob {
 	return &AppWrapper{}
 }
 
-func isAppWrapper(owner *metav1.OwnerReference) bool {
-	return owner.Kind == awv1beta2.AppWrapperKind && strings.HasPrefix(owner.APIVersion, gvk.Group)
-}
-
 func SetupIndexes(ctx context.Context, indexer client.FieldIndexer) error {
 	return jobframework.SetupWorkloadOwnerIndex(ctx, indexer, gvk)
 }
diff --git a/pkg/controller/jobs/deployment/deployment_controller.go b/pkg/controller/jobs/deployment/deployment_controller.go
index 7cf79b819..bc5779bd5 100644
--- a/pkg/controller/jobs/deployment/deployment_controller.go
+++ b/pkg/controller/jobs/deployment/deployment_controller.go
@@ -20,7 +20,6 @@ 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"
@@ -39,21 +38,16 @@ 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"},
-		IsManagingObjectsOwner: isDeployment,
+		SetupIndexes:   SetupIndexes,
+		NewReconciler:  jobframework.NewNoopReconcilerFactory(gvk),
+		GVK:            gvk,
+		SetupWebhook:   SetupWebhook,
+		JobType:        &appsv1.Deployment{},
+		AddToScheme:    appsv1.AddToScheme,
+		DependencyList: []string{"pod"},
 	}))
 }
 
-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
diff --git a/pkg/controller/jobs/job/job_controller.go b/pkg/controller/jobs/job/job_controller.go
index 1bb430de3..2ed476f51 100644
--- a/pkg/controller/jobs/job/job_controller.go
+++ b/pkg/controller/jobs/job/job_controller.go
@@ -59,13 +59,12 @@ const (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupWebhook,
-		JobType:                &batchv1.Job{},
-		IsManagingObjectsOwner: isJob,
-		MultiKueueAdapter:      &multiKueueAdapter{},
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupWebhook,
+		JobType:           &batchv1.Job{},
+		MultiKueueAdapter: &multiKueueAdapter{},
 	}))
 }
 
@@ -88,10 +87,6 @@ var NewReconciler = jobframework.NewGenericReconcilerFactory(NewJob, func(b *bui
 	return b.Watches(&kueue.Workload{}, &parentWorkloadHandler{client: c})
 })
 
-func isJob(owner *metav1.OwnerReference) bool {
-	return owner.Kind == "Job" && owner.APIVersion == gvk.GroupVersion().String()
-}
-
 type parentWorkloadHandler struct {
 	client client.Client
 }
diff --git a/pkg/controller/jobs/jobset/jobset_controller.go b/pkg/controller/jobs/jobset/jobset_controller.go
index acae2f01a..805357a8a 100644
--- a/pkg/controller/jobs/jobset/jobset_controller.go
+++ b/pkg/controller/jobs/jobset/jobset_controller.go
@@ -19,7 +19,6 @@ package jobset
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	batchv1 "k8s.io/api/batch/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
@@ -46,14 +45,13 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupJobSetWebhook,
-		JobType:                &jobsetapi.JobSet{},
-		AddToScheme:            jobsetapi.AddToScheme,
-		IsManagingObjectsOwner: isJobSet,
-		MultiKueueAdapter:      &multiKueueAdapter{},
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupJobSetWebhook,
+		JobType:           &jobsetapi.JobSet{},
+		AddToScheme:       jobsetapi.AddToScheme,
+		MultiKueueAdapter: &multiKueueAdapter{},
 	}))
 }
 
@@ -74,10 +72,6 @@ func NewJob() jobframework.GenericJob {
 
 var NewReconciler = jobframework.NewGenericReconcilerFactory(NewJob)
 
-func isJobSet(owner *metav1.OwnerReference) bool {
-	return owner.Kind == "JobSet" && strings.HasPrefix(owner.APIVersion, "jobset.x-k8s.io/v1")
-}
-
 type JobSet jobsetapi.JobSet
 
 var _ jobframework.GenericJob = (*JobSet)(nil)
diff --git a/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller.go b/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller.go
index a57098f61..87e7514c7 100644
--- a/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller.go
+++ b/pkg/controller/jobs/kubeflow/jobs/paddlejob/paddlejob_controller.go
@@ -19,10 +19,8 @@ package paddlejob
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -50,14 +48,13 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupPaddleJobWebhook,
-		JobType:                &kftraining.PaddleJob{},
-		AddToScheme:            kftraining.AddToScheme,
-		IsManagingObjectsOwner: isPaddleJob,
-		MultiKueueAdapter:      kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupPaddleJobWebhook,
+		JobType:           &kftraining.PaddleJob{},
+		AddToScheme:       kftraining.AddToScheme,
+		MultiKueueAdapter: kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
 	}))
 }
 
@@ -80,10 +77,6 @@ var NewReconciler = jobframework.NewGenericReconcilerFactory(func() jobframework
 	return &kubeflowjob.KubeflowJob{KFJobControl: &JobControl{}}
 })
 
-func isPaddleJob(owner *metav1.OwnerReference) bool {
-	return owner.Kind == kftraining.PaddleJobKind && strings.HasPrefix(owner.APIVersion, kftraining.SchemeGroupVersion.Group)
-}
-
 type JobControl kftraining.PaddleJob
 
 var _ kubeflowjob.KFJobControl = (*JobControl)(nil)
diff --git a/pkg/controller/jobs/kubeflow/jobs/pytorchjob/pytorchjob_controller.go b/pkg/controller/jobs/kubeflow/jobs/pytorchjob/pytorchjob_controller.go
index 889e0e2bb..8c8699466 100644
--- a/pkg/controller/jobs/kubeflow/jobs/pytorchjob/pytorchjob_controller.go
+++ b/pkg/controller/jobs/kubeflow/jobs/pytorchjob/pytorchjob_controller.go
@@ -19,10 +19,8 @@ package pytorchjob
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -50,14 +48,13 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupPyTorchJobWebhook,
-		JobType:                &kftraining.PyTorchJob{},
-		AddToScheme:            kftraining.AddToScheme,
-		IsManagingObjectsOwner: isPyTorchJob,
-		MultiKueueAdapter:      kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupPyTorchJobWebhook,
+		JobType:           &kftraining.PyTorchJob{},
+		AddToScheme:       kftraining.AddToScheme,
+		MultiKueueAdapter: kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
 	}))
 }
 
@@ -80,10 +77,6 @@ var NewReconciler = jobframework.NewGenericReconcilerFactory(func() jobframework
 	return &kubeflowjob.KubeflowJob{KFJobControl: &JobControl{}}
 })
 
-func isPyTorchJob(owner *metav1.OwnerReference) bool {
-	return owner.Kind == kftraining.PyTorchJobKind && strings.HasPrefix(owner.APIVersion, kftraining.SchemeGroupVersion.Group)
-}
-
 type JobControl kftraining.PyTorchJob
 
 var _ kubeflowjob.KFJobControl = (*JobControl)(nil)
diff --git a/pkg/controller/jobs/kubeflow/jobs/tfjob/tfjob_controller.go b/pkg/controller/jobs/kubeflow/jobs/tfjob/tfjob_controller.go
index 916264421..672a800ef 100644
--- a/pkg/controller/jobs/kubeflow/jobs/tfjob/tfjob_controller.go
+++ b/pkg/controller/jobs/kubeflow/jobs/tfjob/tfjob_controller.go
@@ -19,10 +19,8 @@ package tfjob
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -50,14 +48,13 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupTFJobWebhook,
-		JobType:                &kftraining.TFJob{},
-		AddToScheme:            kftraining.AddToScheme,
-		IsManagingObjectsOwner: isTFJob,
-		MultiKueueAdapter:      kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupTFJobWebhook,
+		JobType:           &kftraining.TFJob{},
+		AddToScheme:       kftraining.AddToScheme,
+		MultiKueueAdapter: kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
 	}))
 }
 
@@ -80,10 +77,6 @@ var NewReconciler = jobframework.NewGenericReconcilerFactory(func() jobframework
 	return &kubeflowjob.KubeflowJob{KFJobControl: &JobControl{}}
 })
 
-func isTFJob(owner *metav1.OwnerReference) bool {
-	return owner.Kind == kftraining.TFJobKind && strings.HasPrefix(owner.APIVersion, kftraining.SchemeGroupVersion.Group)
-}
-
 type JobControl kftraining.TFJob
 
 var _ kubeflowjob.KFJobControl = (*JobControl)(nil)
diff --git a/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller.go b/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller.go
index 085d88c35..efb147d80 100644
--- a/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller.go
+++ b/pkg/controller/jobs/kubeflow/jobs/xgboostjob/xgboostjob_controller.go
@@ -19,10 +19,8 @@ package xgboostjob
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -50,14 +48,13 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupXGBoostJobWebhook,
-		JobType:                &kftraining.XGBoostJob{},
-		AddToScheme:            kftraining.AddToScheme,
-		IsManagingObjectsOwner: isXGBoostJob,
-		MultiKueueAdapter:      kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupXGBoostJobWebhook,
+		JobType:           &kftraining.XGBoostJob{},
+		AddToScheme:       kftraining.AddToScheme,
+		MultiKueueAdapter: kubeflowjob.NewMKAdapter(copyJobSpec, copyJobStatus, getEmptyList, gvk, fromObject),
 	}))
 }
 
@@ -80,10 +77,6 @@ var NewReconciler = jobframework.NewGenericReconcilerFactory(func() jobframework
 	return &kubeflowjob.KubeflowJob{KFJobControl: &JobControl{}}
 })
 
-func isXGBoostJob(owner *metav1.OwnerReference) bool {
-	return owner.Kind == kftraining.XGBoostJobKind && strings.HasPrefix(owner.APIVersion, kftraining.SchemeGroupVersion.Group)
-}
-
 type JobControl kftraining.XGBoostJob
 
 var _ kubeflowjob.KFJobControl = (*JobControl)(nil)
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_controller.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_controller.go
index dffce5d29..006da4321 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_controller.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_controller.go
@@ -18,9 +18,7 @@ package leaderworkerset
 
 import (
 	"context"
-	"strings"
 
-	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"
@@ -47,7 +45,6 @@ func init() {
 		JobType:                  &leaderworkersetv1.LeaderWorkerSet{},
 		AddToScheme:              leaderworkersetv1.AddToScheme,
 		DependencyList:           []string{"pod"},
-		IsManagingObjectsOwner:   isLeaderWorkerSet,
 		GVK:                      gvk,
 	}))
 }
@@ -58,10 +55,6 @@ func fromObject(o runtime.Object) *LeaderWorkerSet {
 	return (*LeaderWorkerSet)(o.(*leaderworkersetv1.LeaderWorkerSet))
 }
 
-func isLeaderWorkerSet(ref *metav1.OwnerReference) bool {
-	return ref.Kind == "LeaderWorkerSet" && strings.HasPrefix(ref.APIVersion, "leaderworkerset.x-k8s.io/v1")
-}
-
 func (lws *LeaderWorkerSet) Object() client.Object {
 	return (*leaderworkersetv1.LeaderWorkerSet)(lws)
 }
diff --git a/pkg/controller/jobs/mpijob/mpijob_controller.go b/pkg/controller/jobs/mpijob/mpijob_controller.go
index 2b0c0f95d..e15b26c89 100644
--- a/pkg/controller/jobs/mpijob/mpijob_controller.go
+++ b/pkg/controller/jobs/mpijob/mpijob_controller.go
@@ -19,11 +19,9 @@ package mpijob
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	kfmpi "github.com/kubeflow/mpi-operator/pkg/apis/kubeflow/v2beta1"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -45,14 +43,13 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupMPIJobWebhook,
-		JobType:                &kfmpi.MPIJob{},
-		AddToScheme:            kfmpi.AddToScheme,
-		IsManagingObjectsOwner: isMPIJob,
-		MultiKueueAdapter:      &multiKueueAdapter{},
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupMPIJobWebhook,
+		JobType:           &kfmpi.MPIJob{},
+		AddToScheme:       kfmpi.AddToScheme,
+		MultiKueueAdapter: &multiKueueAdapter{},
 	}))
 }
 
@@ -73,10 +70,6 @@ func NewJob() jobframework.GenericJob {
 
 var NewReconciler = jobframework.NewGenericReconcilerFactory(NewJob)
 
-func isMPIJob(owner *metav1.OwnerReference) bool {
-	return owner.Kind == "MPIJob" && strings.HasPrefix(owner.APIVersion, kfmpi.SchemeGroupVersion.Group)
-}
-
 type MPIJob kfmpi.MPIJob
 
 var _ jobframework.GenericJob = (*MPIJob)(nil)
diff --git a/pkg/controller/jobs/pod/pod_controller.go b/pkg/controller/jobs/pod/pod_controller.go
index 941c0bf49..34c25c7b4 100644
--- a/pkg/controller/jobs/pod/pod_controller.go
+++ b/pkg/controller/jobs/pod/pod_controller.go
@@ -90,20 +90,15 @@ var (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupWebhook,
-		JobType:                &corev1.Pod{},
-		MultiKueueAdapter:      &multiKueueAdapter{},
-		IsManagingObjectsOwner: isPod,
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupWebhook,
+		JobType:           &corev1.Pod{},
+		MultiKueueAdapter: &multiKueueAdapter{},
 	}))
 }
 
-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/raycluster/raycluster_controller.go b/pkg/controller/jobs/raycluster/raycluster_controller.go
index 5da47add1..f4a2f1353 100644
--- a/pkg/controller/jobs/raycluster/raycluster_controller.go
+++ b/pkg/controller/jobs/raycluster/raycluster_controller.go
@@ -19,11 +19,9 @@ package raycluster
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	rayutils "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -48,14 +46,13 @@ const (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupRayClusterWebhook,
-		JobType:                &rayv1.RayCluster{},
-		AddToScheme:            rayv1.AddToScheme,
-		IsManagingObjectsOwner: isRayCluster,
-		MultiKueueAdapter:      &multiKueueAdapter{},
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupRayClusterWebhook,
+		JobType:           &rayv1.RayCluster{},
+		AddToScheme:       rayv1.AddToScheme,
+		MultiKueueAdapter: &multiKueueAdapter{},
 	}))
 }
 
@@ -209,10 +206,6 @@ func GetWorkloadNameForRayCluster(jobName string, jobUID types.UID) string {
 	return jobframework.GetWorkloadNameForOwnerWithGVK(jobName, jobUID, gvk)
 }
 
-func isRayCluster(owner *metav1.OwnerReference) bool {
-	return owner.Kind == "RayCluster" && strings.HasPrefix(owner.APIVersion, "ray.io/v1")
-}
-
 func fromObject(o runtime.Object) *RayCluster {
 	return (*RayCluster)(o.(*rayv1.RayCluster))
 }
diff --git a/pkg/controller/jobs/rayjob/rayjob_controller.go b/pkg/controller/jobs/rayjob/rayjob_controller.go
index 6743045ce..eec0bfef7 100644
--- a/pkg/controller/jobs/rayjob/rayjob_controller.go
+++ b/pkg/controller/jobs/rayjob/rayjob_controller.go
@@ -19,13 +19,11 @@ package rayjob
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	rayutils "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 	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/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -50,14 +48,13 @@ const (
 
 func init() {
 	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
-		SetupIndexes:           SetupIndexes,
-		NewJob:                 NewJob,
-		NewReconciler:          NewReconciler,
-		SetupWebhook:           SetupRayJobWebhook,
-		JobType:                &rayv1.RayJob{},
-		AddToScheme:            rayv1.AddToScheme,
-		IsManagingObjectsOwner: isRayJob,
-		MultiKueueAdapter:      &multiKueueAdapter{},
+		SetupIndexes:      SetupIndexes,
+		NewJob:            NewJob,
+		NewReconciler:     NewReconciler,
+		SetupWebhook:      SetupRayJobWebhook,
+		JobType:           &rayv1.RayJob{},
+		AddToScheme:       rayv1.AddToScheme,
+		MultiKueueAdapter: &multiKueueAdapter{},
 	}))
 }
 
@@ -261,10 +258,6 @@ func GetWorkloadNameForRayJob(jobName string, jobUID types.UID) string {
 	return jobframework.GetWorkloadNameForOwnerWithGVK(jobName, jobUID, gvk)
 }
 
-func isRayJob(owner *metav1.OwnerReference) bool {
-	return owner.Kind == "RayJob" && strings.HasPrefix(owner.APIVersion, "ray.io/v1")
-}
-
 // getSubmitterTemplate returns the PodTemplteSpec of the submitter Job used for RayJob when submissionMode=K8sJobMode
 func getSubmitterTemplate(rayJob *RayJob) *corev1.PodTemplateSpec {
 	if rayJob.Spec.SubmitterPodTemplate != nil {
diff --git a/pkg/controller/jobs/statefulset/statefulset_controller.go b/pkg/controller/jobs/statefulset/statefulset_controller.go
index 502e922dc..0039a747c 100644
--- a/pkg/controller/jobs/statefulset/statefulset_controller.go
+++ b/pkg/controller/jobs/statefulset/statefulset_controller.go
@@ -20,7 +20,6 @@ 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"
@@ -39,21 +38,16 @@ 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,
-		IsManagingObjectsOwner: isStatefulSet,
+		SetupIndexes:   SetupIndexes,
+		NewReconciler:  NewReconciler,
+		SetupWebhook:   SetupWebhook,
+		JobType:        &appsv1.StatefulSet{},
+		AddToScheme:    appsv1.AddToScheme,
+		DependencyList: []string{"pod"},
+		GVK:            gvk,
 	}))
 }
 
-func isStatefulSet(ref *metav1.OwnerReference) bool {
-	return ref.Kind == "StatefulSet" && ref.APIVersion == "apps/v1"
-}
-
 type StatefulSet appsv1.StatefulSet
 
 func fromObject(o runtime.Object) *StatefulSet {



Test patch:
diff --git a/pkg/controller/jobframework/integrationmanager_test.go b/pkg/controller/jobframework/integrationmanager_test.go
index b2c106008..dcfc8be2a 100644
--- a/pkg/controller/jobframework/integrationmanager_test.go
+++ b/pkg/controller/jobframework/integrationmanager_test.go
@@ -365,23 +365,23 @@ func TestGetJobTypeForOwner(t *testing.T) {
 	}
 	manageK1 := func() IntegrationCallbacks {
 		ret := dontManage
-		ret.IsManagingObjectsOwner = func(owner *metav1.OwnerReference) bool { return owner.Kind == "K1" }
-		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K1"}}
+		ret.GVK = schema.GroupVersionKind{Group: "test-group", Version: "v1", Kind: "K1"}
+		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K1", APIVersion: "test-group/v1"}}
 		return ret
 	}()
 	manageK2 := func() IntegrationCallbacks {
 		ret := dontManage
-		ret.IsManagingObjectsOwner = func(owner *metav1.OwnerReference) bool { return owner.Kind == "K2" }
-		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K2"}}
+		ret.GVK = schema.GroupVersionKind{Group: "test-group", Version: "v1", Kind: "K2"}
+		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K2", APIVersion: "test-group/v1"}}
 		return ret
 	}()
 	externalK3 := func() runtime.Object {
-		return &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K3"}}
+		return &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K3", APIVersion: "test-group/v1"}}
 	}()
 	disabledK4 := func() IntegrationCallbacks {
 		ret := dontManage
-		ret.IsManagingObjectsOwner = func(owner *metav1.OwnerReference) bool { return owner.Kind == "K4" }
-		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K4"}}
+		ret.GVK = schema.GroupVersionKind{Group: "test-group", Version: "v1", Kind: "K4"}
+		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K4", APIVersion: "test-group/v1"}}
 		return ret
 	}()
 
@@ -406,23 +406,23 @@ func TestGetJobTypeForOwner(t *testing.T) {
 		wantJobType runtime.Object
 	}{
 		"K1": {
-			owner:       &metav1.OwnerReference{Kind: "K1"},
+			owner:       &metav1.OwnerReference{Kind: "K1", APIVersion: "test-group/v1"},
 			wantJobType: manageK1.JobType,
 		},
 		"K2": {
-			owner:       &metav1.OwnerReference{Kind: "K2"},
+			owner:       &metav1.OwnerReference{Kind: "K2", APIVersion: "test-group/v1"},
 			wantJobType: manageK2.JobType,
 		},
 		"K3": {
-			owner:       &metav1.OwnerReference{Kind: "K3"},
+			owner:       &metav1.OwnerReference{Kind: "K3", APIVersion: "test-group/v1"},
 			wantJobType: externalK3,
 		},
 		"K4": {
-			owner:       &metav1.OwnerReference{Kind: "K4"},
+			owner:       &metav1.OwnerReference{Kind: "K4", APIVersion: "test-group/v1"},
 			wantJobType: nil,
 		},
 		"K5": {
-			owner:       &metav1.OwnerReference{Kind: "K5"},
+			owner:       &metav1.OwnerReference{Kind: "K5", APIVersion: "test-group/v1"},
 			wantJobType: nil,
 		},
 	}
@@ -509,38 +509,29 @@ func TestOwnerFrameworkEnabledChecks(t *testing.T) {
 	}
 	manageK1 := func() IntegrationCallbacks {
 		ret := dontManage
-		ret.IsManagingObjectsOwner = func(owner *metav1.OwnerReference) bool { return owner.Kind == "K1" }
-		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K1"}}
-		ret.GVK = ret.JobType.GetObjectKind().GroupVersionKind()
+		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K1", APIVersion: "test-group/v1"}}
+		ret.GVK = schema.GroupVersionKind{Group: "test-group", Version: "v1", Kind: "K1"}
 		return ret
 	}()
-	manageK2 := func() IntegrationCallbacks {
-		ret := dontManage
-		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K2"}}
-		ret.GVK = ret.JobType.GetObjectKind().GroupVersionKind()
-		return ret
-	}()
-	externalK3 := func() runtime.Object {
-		return &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K3"}}
+	externalK2 := func() runtime.Object {
+		return &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K2", APIVersion: "test-group/v1"}}
 	}()
-	disabledK4 := func() IntegrationCallbacks {
+	disabledK3 := func() IntegrationCallbacks {
 		ret := dontManage
-		ret.IsManagingObjectsOwner = func(owner *metav1.OwnerReference) bool { return owner.Kind == "K4" }
-		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K4"}}
-		ret.GVK = ret.JobType.GetObjectKind().GroupVersionKind()
+		ret.JobType = &metav1.PartialObjectMetadata{TypeMeta: metav1.TypeMeta{Kind: "K3", APIVersion: "test-group/v1"}}
+		ret.GVK = schema.GroupVersionKind{Group: "test-group", Version: "v1", Kind: "K3"}
 		return ret
 	}()
 
 	mgr := integrationManager{
-		names: []string{"manageK1", "dontManage", "manageK2", "disabledK4"},
+		names: []string{"dontManage", "manageK1", "disabledK3"},
 		integrations: map[string]IntegrationCallbacks{
 			"dontManage": dontManage,
 			"manageK1":   manageK1,
-			"manageK2":   manageK2,
-			"disabledK4": disabledK4,
+			"disabledK3": disabledK3,
 		},
 		externalIntegrations: map[string]runtime.Object{
-			"externalK3": externalK3,
+			"externalK2": externalK2,
 		},
 	}
 	mgr.enableIntegration("manageK1")
@@ -552,27 +543,22 @@ func TestOwnerFrameworkEnabledChecks(t *testing.T) {
 		wantOwnerIsEnabled bool
 	}{
 		"K1": {
-			owner:              &metav1.OwnerReference{Kind: "K1"},
+			owner:              &metav1.OwnerReference{Kind: "K1", APIVersion: "test-group/v1"},
 			wantOwnerIsManaged: true,
 			wantOwnerIsEnabled: true,
 		},
 		"K2": {
-			owner:              &metav1.OwnerReference{Kind: "K2"},
-			wantOwnerIsManaged: false,
-			wantOwnerIsEnabled: true,
-		},
-		"K3": {
-			owner:              &metav1.OwnerReference{Kind: "K3"},
+			owner:              &metav1.OwnerReference{Kind: "K2", APIVersion: "test-group/v1"},
 			wantOwnerIsManaged: true,
 			wantOwnerIsEnabled: true,
 		},
-		"K4": {
-			owner:              &metav1.OwnerReference{Kind: "K4"},
+		"K3": {
+			owner:              &metav1.OwnerReference{Kind: "K3", APIVersion: "test-group/v1"},
 			wantOwnerIsManaged: false,
 			wantOwnerIsEnabled: false,
 		},
 		"K5": {
-			owner:              &metav1.OwnerReference{Kind: "K5"},
+			owner:              &metav1.OwnerReference{Kind: "K5", APIVersion: "test-group/v1"},
 			wantOwnerIsManaged: false,
 			wantOwnerIsEnabled: false,
 		},