Repository: kubernetes-sigs/kueue

Issue:
Support KubeRay in MultiKueue via managedBy
**What would you like to be added**:

Support for KubeRay via managedBy in MultiKueue.

The relevant support for the managedBy field has been recently merged in KubeRay, see https://github.com/ray-project/kuberay/issues/2544, and will be released most likely in 1.3.
Until then we can use the main branch of kuberay in Kueue to test it all works. Once KubeRay is released we can switch to the released version and merge to Kueue.

**Why is this needed**:

Support of KubeRay via managedBy in MultiKueue, allowing for:
- ease of setup where full KubeRay can be installed on the management cluster (not just CRDs)
- hybrid deployments, where some RayJobs are executed via MultiKueue and some locally on the management cluster


Hints from issue comments:
@mimowo: /assign @mszadkow 

cc @mbobrovskyi @andrewsykim @mwielgus @mwysokin @mszadkow: This ticket requires to first provide RayJob and RayCluster Multikueue adapter and setup e2e tests,
as we don't have them yet.
First PR will be about adapter and multikueue e2e tests with ray-operator v1.2.2- no managedBy yet, as this requires `latest`.
Second PR will be the attempt to work with ray-operator `latest` and use of managedBy.
   @mimowo: sgtm, we can start with tests using 1.2.2 but without managedBy field and merge it as a starter. Or alternatively already use latest (main) of KubeRay just for testing purposes - we will merge once KubeRay is released.@mszadkow: Another problem that I have right now is that Kuberay clusters startup time is huge for e2e tests.
It's about 5 minutes, maybe it's due to requirement to use rayproject image, but setting anything else as image results in job to stall at initialisation.
Normally we use `sleep` image just to call anything, but it won't work with ray cluster.@mimowo: Ok, in that case we may need to have a separate CI for Ray. 

However, let me first understand what you exactly mean by "cluster startup time" - is this the installation of the KubeRay, or time to run the first RayJob? Does it also take long to run follow-up Jobs?

Also, please make sure you are rebased against the main branch, because recently we increased CPU limits for Kueue to 2000m which might be relevant here too.@andrewsykim: > Normally we use sleep image just to call anything, but it won't work with ray cluster.

It may be possible to construct a dummy RayCluster that just calls `sleep` with a lighter image, but I haven't tried this myself. Assuming the start-up time issue is due to pulling the default Ray images @mimowo: thank you @andrewsykim for the suggestion. @mszadkow can we try that? maybe you already did and hit some complications?@mszadkow: Well I have tried to load the image up-front to the test cluster (kind), but then I had space limitation issues, at least locally.
Will have to verify what happens in CI.
@andrewsykim did you mean I could build lighter version of `rayproject/ray` ?@andrewsykim: > @andrewsykim did you mean I could build lighter version of rayproject/ray ?

No, when you constuct a RayJob or RayCluster, you can pass arbitrary images for the Head and Worker pods. So you can try using much smaller images used in other tests that just need to run `sleep`. @mszadkow: I found a way to run other type of image, just with bash and calling sleep.
However it seems that kuberay overrides KUBERAY_GEN_RAY_START_CMD and never mind what I put there it's ` ray start --head  --dashboard-host=0.0.0.0  --metrics-export-port=8080  --block  --dashboard-agent-listen-port=52365`.
Then I would have to install ray to the docker...
Let's first go with `rayproject` image@mszadkow: I think, I will start early integration of v1.3.0-rc.0 within days
https://github.com/ray-project/kuberay/releases/tag/v1.3.0-rc.0
quay.io/kuberay/operator:v1.3.0-rc.0
@mszadkow: @andrewsykim @mimowo 
I have encountered an issue with kuberay v1.3.0-rc.0 upgrade.

Kuberay has [k8s.io/component-base](http://k8s.io/component-base) v0.29.6
Kueue has [k8s.io/component-base](http://k8s.io/component-base) v0.32.1

There is a small change that affect us https://github.com/kubernetes/component-base/blob/264c1fd30132a3b36b7588e50ac54eb0ff75f26a/featuregate/testing/feature_gate.go#L47
this function now returns nothing, while before it returned a cleanup function.

This result in:
vendor/github.com/ray-project/kuberay/ray-operator/pkg/features/features.go:40:9: featuregatetesting.SetFeatureGateDuringTest(tb, utilfeature.DefaultFeatureGate, f, value) (no value) used as value
make: *** [build] Error 1

Question remains if it's ok to upgrade Kuberay's component base to v0.32.1 for 1.3.0 release?@andrewsykim: No objections to upgrade component base to v0.32.1 -- do you mind opening a PR for that?@mszadkow: I am in the process of doing that, though I am afraid it won't just this package.
But let's see the results first ;) @mszadkow: @andrewsykim @mimowo 
After lengthy battle here is the PR: https://github.com/ray-project/kuberay/pull/3004@mszadkow: @andrewsykim is there any slack channel on `kubernetes.slack.com` dedicated to kuberay?
I only found dedicated workspace: https://github.com/ray-project/kuberay/issues/1059
@tenzen-y: > [@andrewsykim](https://github.com/andrewsykim) is there any slack channel on `kubernetes.slack.com` dedicated to kuberay? I only found dedicated workspace: [ray-project/kuberay#1059](https://github.com/ray-project/kuberay/issues/1059)

IIUC, Ray slack is only ray.slack.com
@mszadkow: @mimowo @tenzen-y @andrewsykim 
I have tested locally that [kuberay release-1.3 with backport](https://github.com/ray-project/kuberay/pull/3163) works with Kueue.
I even run MultiKueue e2e tests with managedBy and kuberay-operator running on manager cluster and it passes, thus I think it is safe to proceed whenever Kuberay releases next 1.3.X version.@tenzen-y: > [@mimowo](https://github.com/mimowo) [@tenzen-y](https://github.com/tenzen-y) [@andrewsykim](https://github.com/andrewsykim) I have tested locally that [kuberay release-1.3 with backport](https://github.com/ray-project/kuberay/pull/3163) works with Kueue. I even run MultiKueue e2e tests with managedBy and kuberay-operator running on manager cluster and it passes, thus I think it is safe to proceed whenever Kuberay releases next 1.3.X version.

Thank you for your effort!
@mimowo: /reopen
this is not done yet, only KubeRay is bumped so far. which is the prerequisite@k8s-ci-robot: @mimowo: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/3822#issuecomment-2733406446):

>/reopen
>this is not done yet, only KubeRay is bumped so far. which is the prerequisite


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@mszadkow: This https://github.com/kubernetes-sigs/kueue/pull/4677 should conclude the task.@mimowo: We ahould also follow up with docs update. but this can be post release@mimowo: /release-note-edit
```release-note
Support KubeRay (RayJob and RayCluster integrations) in MultiKueue via managedBy. 
This allows using the integrations and installing the KubeRay operator on the management cluster.
```
@mszadkow: @mimowo Docs have beed updated - https://github.com/kubernetes-sigs/kueue/pull/4686.
Are there any other places to be updated or we could close this one?@mimowo: /close@k8s-ci-robot: @mimowo: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/3822#issuecomment-2809330402):

>/close


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

Solution patch:
diff --git a/hack/e2e-common.sh b/hack/e2e-common.sh
index 99a4e2156..ef9b01ef3 100644
--- a/hack/e2e-common.sh
+++ b/hack/e2e-common.sh
@@ -49,7 +49,6 @@ if [[ -n ${KUBERAY_VERSION:-} ]]; then
     export KUBERAY_IMAGE=quay.io/kuberay/operator:${KUBERAY_VERSION}
     export KUBERAY_RAY_IMAGE=rayproject/ray:2.9.0
     export KUBERAY_RAY_IMAGE_ARM=rayproject/ray:2.9.0-aarch64
-    export KUBERAY_CRDS=${ROOT_DIR}/dep-crds/ray-operator/crd/bases
 fi
 
 if [[ -n ${LEADERWORKERSET_VERSION:-} ]]; then
diff --git a/hack/multikueue-e2e-test.sh b/hack/multikueue-e2e-test.sh
index d143199e5..8df034dce 100755
--- a/hack/multikueue-e2e-test.sh
+++ b/hack/multikueue-e2e-test.sh
@@ -100,9 +100,7 @@ function kind_load {
     install_mpi "$WORKER2_KIND_CLUSTER_NAME"
 
     ## KUBERAY
-    kubectl config use-context "kind-${MANAGER_KIND_CLUSTER_NAME}"
-    kubectl apply --server-side -f "${KUBERAY_CRDS}"
-
+    install_kuberay "$MANAGER_KIND_CLUSTER_NAME"
     install_kuberay "$WORKER1_KIND_CLUSTER_NAME"
     install_kuberay "$WORKER2_KIND_CLUSTER_NAME"
 }
diff --git a/pkg/controller/jobs/raycluster/raycluster_controller.go b/pkg/controller/jobs/raycluster/raycluster_controller.go
index 00765942f..dbc61b013 100644
--- a/pkg/controller/jobs/raycluster/raycluster_controller.go
+++ b/pkg/controller/jobs/raycluster/raycluster_controller.go
@@ -33,6 +33,7 @@ import (
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/podset"
 )
 
@@ -77,6 +78,7 @@ var NewReconciler = jobframework.NewGenericReconcilerFactory(NewJob)
 type RayCluster rayv1.RayCluster
 
 var _ jobframework.GenericJob = (*RayCluster)(nil)
+var _ jobframework.JobWithManagedBy = (*RayCluster)(nil)
 
 func (j *RayCluster) Object() client.Object {
 	return (*rayv1.RayCluster)(j)
@@ -203,3 +205,17 @@ func isRayCluster(owner *metav1.OwnerReference) bool {
 func fromObject(o runtime.Object) *RayCluster {
 	return (*RayCluster)(o.(*rayv1.RayCluster))
 }
+
+func (j *RayCluster) CanDefaultManagedBy() bool {
+	jobSpecManagedBy := j.Spec.ManagedBy
+	return features.Enabled(features.MultiKueue) &&
+		(jobSpecManagedBy == nil || *jobSpecManagedBy == rayutils.KubeRayController)
+}
+
+func (j *RayCluster) ManagedBy() *string {
+	return j.Spec.ManagedBy
+}
+
+func (j *RayCluster) SetManagedBy(managedBy *string) {
+	j.Spec.ManagedBy = managedBy
+}
diff --git a/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter.go b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter.go
index 4ad2b19f2..2ba0fe3e8 100644
--- a/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter.go
+++ b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -81,6 +82,9 @@ func (b *multiKueueAdapter) SyncJob(ctx context.Context, localClient client.Clie
 	remoteJob.Labels[constants.PrebuiltWorkloadLabel] = workloadName
 	remoteJob.Labels[kueue.MultiKueueOriginLabel] = origin
 
+	// clear the managedBy enables the controller to take over
+	remoteJob.Spec.ManagedBy = nil
+
 	return remoteClient.Create(ctx, &remoteJob)
 }
 
@@ -96,6 +100,15 @@ func (b *multiKueueAdapter) KeepAdmissionCheckPending() bool {
 }
 
 func (b *multiKueueAdapter) IsJobManagedByKueue(ctx context.Context, c client.Client, key types.NamespacedName) (bool, string, error) {
+	job := rayv1.RayCluster{}
+	err := c.Get(ctx, key, &job)
+	if err != nil {
+		return false, "", err
+	}
+	jobControllerName := ptr.Deref(job.Spec.ManagedBy, "")
+	if jobControllerName != kueue.MultiKueueControllerName {
+		return false, fmt.Sprintf("Expecting spec.managedBy to be %q not %q", kueue.MultiKueueControllerName, jobControllerName), nil
+	}
 	return true, "", nil
 }
 
diff --git a/pkg/controller/jobs/raycluster/raycluster_webhook.go b/pkg/controller/jobs/raycluster/raycluster_webhook.go
index d2201e1be..7df0b0b9a 100644
--- a/pkg/controller/jobs/raycluster/raycluster_webhook.go
+++ b/pkg/controller/jobs/raycluster/raycluster_webhook.go
@@ -29,6 +29,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
+	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework/webhook"
 	"sigs.k8s.io/kueue/pkg/queue"
@@ -45,6 +46,7 @@ type RayClusterWebhook struct {
 	queues                       *queue.Manager
 	manageJobsWithoutQueueName   bool
 	managedJobsNamespaceSelector labels.Selector
+	cache                        *cache.Cache
 }
 
 // SetupRayClusterWebhook configures the webhook for rayv1 RayCluster.
@@ -58,6 +60,7 @@ func SetupRayClusterWebhook(mgr ctrl.Manager, opts ...jobframework.Option) error
 		queues:                       options.Queues,
 		manageJobsWithoutQueueName:   options.ManageJobsWithoutQueueName,
 		managedJobsNamespaceSelector: options.ManagedJobsNamespaceSelector,
+		cache:                        options.Cache,
 	}
 	obj := &rayv1.RayCluster{}
 	return webhook.WebhookManagedBy(mgr).
@@ -77,7 +80,11 @@ func (w *RayClusterWebhook) Default(ctx context.Context, obj runtime.Object) err
 	log := ctrl.LoggerFrom(ctx).WithName("raycluster-webhook")
 	log.V(10).Info("Applying defaults")
 	jobframework.ApplyDefaultLocalQueue(job.Object(), w.queues.DefaultLocalQueueExist)
-	return jobframework.ApplyDefaultForSuspend(ctx, job, w.client, w.manageJobsWithoutQueueName, w.managedJobsNamespaceSelector)
+	if err := jobframework.ApplyDefaultForSuspend(ctx, job, w.client, w.manageJobsWithoutQueueName, w.managedJobsNamespaceSelector); err != nil {
+		return err
+	}
+	jobframework.ApplyDefaultForManagedBy(job, w.queues, w.cache, log)
+	return nil
 }
 
 // +kubebuilder:webhook:path=/validate-ray-io-v1-raycluster,mutating=false,failurePolicy=fail,sideEffects=None,groups=ray.io,resources=rayclusters,verbs=create;update,versions=v1,name=vraycluster.kb.io,admissionReviewVersions=v1
diff --git a/pkg/controller/jobs/rayjob/rayjob_controller.go b/pkg/controller/jobs/rayjob/rayjob_controller.go
index 9e13fd47b..f29bf452b 100644
--- a/pkg/controller/jobs/rayjob/rayjob_controller.go
+++ b/pkg/controller/jobs/rayjob/rayjob_controller.go
@@ -34,6 +34,7 @@ import (
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/podset"
 )
 
@@ -79,6 +80,7 @@ var NewReconciler = jobframework.NewGenericReconcilerFactory(NewJob)
 type RayJob rayv1.RayJob
 
 var _ jobframework.GenericJob = (*RayJob)(nil)
+var _ jobframework.JobWithManagedBy = (*RayJob)(nil)
 
 func (j *RayJob) Object() client.Object {
 	return (*rayv1.RayJob)(j)
@@ -283,3 +285,17 @@ func getSubmitterTemplate(rayJob *RayJob) *corev1.PodTemplateSpec {
 		},
 	}
 }
+
+func (j *RayJob) CanDefaultManagedBy() bool {
+	jobSpecManagedBy := j.Spec.ManagedBy
+	return features.Enabled(features.MultiKueue) &&
+		(jobSpecManagedBy == nil || *jobSpecManagedBy == rayutils.KubeRayController)
+}
+
+func (j *RayJob) ManagedBy() *string {
+	return j.Spec.ManagedBy
+}
+
+func (j *RayJob) SetManagedBy(managedBy *string) {
+	j.Spec.ManagedBy = managedBy
+}
diff --git a/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter.go b/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter.go
index 5b84bea11..cfed41151 100644
--- a/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter.go
+++ b/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -80,6 +81,9 @@ func (b *multiKueueAdapter) SyncJob(ctx context.Context, localClient client.Clie
 	remoteJob.Labels[constants.PrebuiltWorkloadLabel] = workloadName
 	remoteJob.Labels[kueue.MultiKueueOriginLabel] = origin
 
+	// clear the managedBy enables the controller to take over
+	remoteJob.Spec.ManagedBy = nil
+
 	return remoteClient.Create(ctx, &remoteJob)
 }
 
@@ -95,6 +99,15 @@ func (b *multiKueueAdapter) KeepAdmissionCheckPending() bool {
 }
 
 func (b *multiKueueAdapter) IsJobManagedByKueue(ctx context.Context, c client.Client, key types.NamespacedName) (bool, string, error) {
+	job := rayv1.RayJob{}
+	err := c.Get(ctx, key, &job)
+	if err != nil {
+		return false, "", err
+	}
+	jobControllerName := ptr.Deref(job.Spec.ManagedBy, "")
+	if jobControllerName != kueue.MultiKueueControllerName {
+		return false, fmt.Sprintf("Expecting spec.managedBy to be %q not %q", kueue.MultiKueueControllerName, jobControllerName), nil
+	}
 	return true, "", nil
 }
 
diff --git a/pkg/controller/jobs/rayjob/rayjob_webhook.go b/pkg/controller/jobs/rayjob/rayjob_webhook.go
index 582202f5b..eeb11107a 100644
--- a/pkg/controller/jobs/rayjob/rayjob_webhook.go
+++ b/pkg/controller/jobs/rayjob/rayjob_webhook.go
@@ -29,6 +29,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
+	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework/webhook"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -46,6 +47,7 @@ type RayJobWebhook struct {
 	queues                       *queue.Manager
 	manageJobsWithoutQueueName   bool
 	managedJobsNamespaceSelector labels.Selector
+	cache                        *cache.Cache
 }
 
 // SetupRayJobWebhook configures the webhook for RayJob.
@@ -56,6 +58,7 @@ func SetupRayJobWebhook(mgr ctrl.Manager, opts ...jobframework.Option) error {
 		queues:                       options.Queues,
 		manageJobsWithoutQueueName:   options.ManageJobsWithoutQueueName,
 		managedJobsNamespaceSelector: options.ManagedJobsNamespaceSelector,
+		cache:                        options.Cache,
 	}
 	obj := &rayv1.RayJob{}
 	return webhook.WebhookManagedBy(mgr).
@@ -71,11 +74,15 @@ var _ admission.CustomDefaulter = &RayJobWebhook{}
 
 // Default implements webhook.CustomDefaulter so a webhook will be registered for the type
 func (w *RayJobWebhook) Default(ctx context.Context, obj runtime.Object) error {
-	job := obj.(*rayv1.RayJob)
+	job := fromObject(obj)
 	log := ctrl.LoggerFrom(ctx).WithName("rayjob-webhook")
 	log.V(5).Info("Applying defaults")
-	jobframework.ApplyDefaultLocalQueue((*RayJob)(job).Object(), w.queues.DefaultLocalQueueExist)
-	return jobframework.ApplyDefaultForSuspend(ctx, (*RayJob)(job), w.client, w.manageJobsWithoutQueueName, w.managedJobsNamespaceSelector)
+	jobframework.ApplyDefaultLocalQueue(job.Object(), w.queues.DefaultLocalQueueExist)
+	if err := jobframework.ApplyDefaultForSuspend(ctx, job, w.client, w.manageJobsWithoutQueueName, w.managedJobsNamespaceSelector); err != nil {
+		return err
+	}
+	jobframework.ApplyDefaultForManagedBy(job, w.queues, w.cache, log)
+	return nil
 }
 
 // +kubebuilder:webhook:path=/validate-ray-io-v1-rayjob,mutating=false,failurePolicy=fail,sideEffects=None,groups=ray.io,resources=rayjobs,verbs=create;update,versions=v1,name=vrayjob.kb.io,admissionReviewVersions=v1



Test patch:
diff --git a/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter_test.go b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter_test.go
index 7b9d23f59..e37ffc1a0 100644
--- a/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter_test.go
+++ b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter_test.go
@@ -18,6 +18,7 @@ package raycluster
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -146,6 +147,51 @@ func TestMultiKueueAdapter(t *testing.T) {
 				return adapter.DeleteRemoteObject(ctx, workerClient, types.NamespacedName{Name: "raycluster1", Namespace: TestNamespace})
 			},
 		},
+		"raycluster with wrong managedBy is not considered managed": {
+			managersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					ManagedBy("some-other-controller").
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "raycluster1", Namespace: TestNamespace}); isManged {
+					return errors.New("expecting false")
+				}
+				return nil
+			},
+			wantManagersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					ManagedBy("some-other-controller").
+					Obj(),
+			},
+		},
+
+		"raycluster managedBy multikueue": {
+			managersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					ManagedBy(kueue.MultiKueueControllerName).
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "raycluster1", Namespace: TestNamespace}); !isManged {
+					return errors.New("expecting true")
+				}
+				return nil
+			},
+			wantManagersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					ManagedBy(kueue.MultiKueueControllerName).
+					Obj(),
+			},
+		},
+		"missing raycluster is not considered managed": {
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "raycluster1", Namespace: TestNamespace}); isManged {
+					return errors.New("expecting false")
+				}
+				return nil
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
diff --git a/pkg/controller/jobs/raycluster/raycluster_webhook_test.go b/pkg/controller/jobs/raycluster/raycluster_webhook_test.go
index 2c55411ef..c6f4eda22 100644
--- a/pkg/controller/jobs/raycluster/raycluster_webhook_test.go
+++ b/pkg/controller/jobs/raycluster/raycluster_webhook_test.go
@@ -122,6 +122,7 @@ func TestValidateDefault(t *testing.T) {
 			wh := &RayClusterWebhook{
 				manageJobsWithoutQueueName: tc.manageAll,
 				queues:                     queueManager,
+				cache:                      cqCache,
 			}
 			result := tc.oldJob.DeepCopy()
 			if err := wh.Default(context.Background(), result); err != nil {
diff --git a/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter_test.go b/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter_test.go
index bd01d505e..ef64aaf6c 100644
--- a/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter_test.go
+++ b/pkg/controller/jobs/rayjob/rayjob_multikueue_adapter_test.go
@@ -18,6 +18,7 @@ package rayjob
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -145,6 +146,51 @@ func TestMultiKueueAdapter(t *testing.T) {
 				return adapter.DeleteRemoteObject(ctx, workerClient, types.NamespacedName{Name: "rayjob1", Namespace: TestNamespace})
 			},
 		},
+		"job with wrong managedBy is not considered managed": {
+			managersRayJobs: []rayv1.RayJob{
+				*rayJobBuilder.Clone().
+					ManagedBy("some-other-controller").
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "rayjob1", Namespace: TestNamespace}); isManged {
+					return errors.New("expecting false")
+				}
+				return nil
+			},
+			wantManagersRayJobs: []rayv1.RayJob{
+				*rayJobBuilder.Clone().
+					ManagedBy("some-other-controller").
+					Obj(),
+			},
+		},
+
+		"job managedBy multikueue": {
+			managersRayJobs: []rayv1.RayJob{
+				*rayJobBuilder.Clone().
+					ManagedBy(kueue.MultiKueueControllerName).
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "rayjob1", Namespace: TestNamespace}); !isManged {
+					return errors.New("expecting true")
+				}
+				return nil
+			},
+			wantManagersRayJobs: []rayv1.RayJob{
+				*rayJobBuilder.Clone().
+					ManagedBy(kueue.MultiKueueControllerName).
+					Obj(),
+			},
+		},
+		"missing job is not considered managed": {
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "rayjob1", Namespace: TestNamespace}); isManged {
+					return errors.New("expecting false")
+				}
+				return nil
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
diff --git a/pkg/controller/jobs/rayjob/rayjob_webhook_test.go b/pkg/controller/jobs/rayjob/rayjob_webhook_test.go
index e34c6ac55..d95588642 100644
--- a/pkg/controller/jobs/rayjob/rayjob_webhook_test.go
+++ b/pkg/controller/jobs/rayjob/rayjob_webhook_test.go
@@ -121,6 +121,7 @@ func TestValidateDefault(t *testing.T) {
 			wh := &RayJobWebhook{
 				manageJobsWithoutQueueName: tc.manageAll,
 				queues:                     queueManager,
+				cache:                      cqCache,
 			}
 			result := tc.oldJob.DeepCopy()
 			if err := wh.Default(context.Background(), result); err != nil {
diff --git a/pkg/util/testingjobs/raycluster/wrappers.go b/pkg/util/testingjobs/raycluster/wrappers.go
index e47f73254..088f10263 100644
--- a/pkg/util/testingjobs/raycluster/wrappers.go
+++ b/pkg/util/testingjobs/raycluster/wrappers.go
@@ -195,8 +195,8 @@ func (j *ClusterWrapper) StatusConditions(c metav1.Condition) *ClusterWrapper {
 }
 
 // ManagedBy adds a managedby.
-func (j *ClusterWrapper) ManagedBy(_ string) *ClusterWrapper {
-	// j.Spec.ManagedBy = &c
+func (j *ClusterWrapper) ManagedBy(c string) *ClusterWrapper {
+	j.Spec.ManagedBy = &c
 	return j
 }
 
diff --git a/pkg/util/testingjobs/rayjob/wrappers.go b/pkg/util/testingjobs/rayjob/wrappers.go
index a0f3e0f25..479d1abbb 100644
--- a/pkg/util/testingjobs/rayjob/wrappers.go
+++ b/pkg/util/testingjobs/rayjob/wrappers.go
@@ -279,3 +279,8 @@ func (j *JobWrapper) Env(rayType rayv1.RayNodeType, name, value string) *JobWrap
 	}
 	return j
 }
+
+func (j *JobWrapper) ManagedBy(c string) *JobWrapper {
+	j.Spec.ManagedBy = &c
+	return j
+}
diff --git a/test/e2e/multikueue/suite_test.go b/test/e2e/multikueue/suite_test.go
index 273cef599..5c1344562 100644
--- a/test/e2e/multikueue/suite_test.go
+++ b/test/e2e/multikueue/suite_test.go
@@ -297,6 +297,7 @@ var _ = ginkgo.BeforeSuite(func() {
 	util.WaitForAppWrapperAvailability(ctx, k8sWorker1Client)
 	util.WaitForAppWrapperAvailability(ctx, k8sWorker2Client)
 
+	util.WaitForKubeRayOperatorAvailability(ctx, k8sManagerClient)
 	util.WaitForKubeRayOperatorAvailability(ctx, k8sWorker1Client)
 	util.WaitForKubeRayOperatorAvailability(ctx, k8sWorker2Client)