Repository: kubernetes-sigs/kueue

Issue:
Graduate MultiKueueBatchJobWithManagedBy to beta
https://github.com/kubernetes-sigs/kueue/blob/834c71daa68ee95a3b89b87d473793e0929645bd/pkg/features/kube_features.go#L84

Looking at our feature gates, I see this MultiKueue feature is still alpha.

Should we graduate this feature to beta?

I think we still want to graduate this feature to beta but it relies on ManagedBy being enabled in all supported versions of Kubernetes.

Looking at [ManagedBy KEP](https://github.com/kubernetes/enhancements/issues/4368), I think we can promote this to beta when 1.31 is out of support. Looking at the Kubernetes release scheduler, that should be around the 0.15 timeframe of Kueue.

1.31 will be out of support by end of October so we can graduate this feature to beta once that happens.

WDYT @tenzen-y @gabesaba?

Discussion:
@tenzen-y: > Looking at https://github.com/kubernetes/enhancements/issues/4368, I think we can promote this to beta when 1.31 is out of support. Looking at the Kubernetes release scheduler, that should be around the 0.15 timeframe of Kueue.

Yes, we can work on that in the v0.15 release cycle.
@mimowo: This is ready to be taken in 0.15 I think@mimowo: @kannon92 would you like to work on it?@kannon92: /assign

Solution patch:
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index 230badd32..6056a64e5 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -240,6 +240,7 @@ var defaultVersionedFeatureGates = map[featuregate.Feature]featuregate.Versioned
 	},
 	MultiKueueBatchJobWithManagedBy: {
 		{Version: version.MustParse("0.8"), Default: false, PreRelease: featuregate.Alpha},
+		{Version: version.MustParse("0.15"), Default: true, PreRelease: featuregate.Beta},
 	},
 	TopologyAwareScheduling: {
 		{Version: version.MustParse("0.9"), Default: false, PreRelease: featuregate.Alpha},
diff --git a/site/content/en/docs/installation/_index.md b/site/content/en/docs/installation/_index.md
index 03c2da36d..529662891 100644
--- a/site/content/en/docs/installation/_index.md
+++ b/site/content/en/docs/installation/_index.md
@@ -268,7 +268,8 @@ spec:
 | `FlavorFungibility`                           | `true`  | Beta  | 0.5   |       |
 | `MultiKueue`                                  | `false` | Alpha | 0.6   | 0.8   |
 | `MultiKueue`                                  | `true`  | Beta  | 0.9   |       |
-| `MultiKueueBatchJobWithManagedBy`             | `false` | Alpha | 0.8   |       |
+| `MultiKueueBatchJobWithManagedBy`             | `false` | Alpha | 0.8   | 0.15  |
+| `MultiKueueBatchJobWithManagedBy`             | `true`  | Beta  | 0.15    |       |
 | `PartialAdmission`                            | `false` | Alpha | 0.4   | 0.4   |
 | `PartialAdmission`                            | `true`  | Beta  | 0.5   |       |
 | `VisibilityOnDemand`                          | `false` | Alpha | 0.6   | 0.8   |



Test patch:
diff --git a/pkg/controller/admissionchecks/multikueue/workload_test.go b/pkg/controller/admissionchecks/multikueue/workload_test.go
index e389aaaf1..d23208c85 100644
--- a/pkg/controller/admissionchecks/multikueue/workload_test.go
+++ b/pkg/controller/admissionchecks/multikueue/workload_test.go
@@ -299,6 +299,7 @@ func TestWlReconcile(t *testing.T) {
 			},
 		},
 		"wl without reservation, clears the workload objects (withoutJobManagedBy)": {
+			features:     map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
 			reconcileFor: "wl1",
 			managersJobs: []batchv1.Job{*baseJobBuilder.Clone().Obj()},
 			managersWorkloads: []kueue.Workload{
@@ -501,6 +502,7 @@ func TestWlReconcile(t *testing.T) {
 			},
 		},
 		"remote wl with reservation (withoutJobManagedBy)": {
+			features:     map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
 			reconcileFor: "wl1",
 			managersWorkloads: []kueue.Workload{
 				*baseWorkloadBuilder.Clone().
@@ -565,6 +567,7 @@ func TestWlReconcile(t *testing.T) {
 			},
 		},
 		"remote wl with reservation (withoutJobManagedBy, MultiKueueDispatcherModeIncremental)": {
+			features:       map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
 			reconcileFor:   "wl1",
 			dispatcherName: ptr.To(config.MultiKueueDispatcherModeIncremental),
 			managersWorkloads: []kueue.Workload{
@@ -702,6 +705,7 @@ func TestWlReconcile(t *testing.T) {
 			},
 		},
 		"remote job is changing status, the local job is not updated (withoutJobManagedBy)": {
+			features:     map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
 			reconcileFor: "wl1",
 			managersWorkloads: []kueue.Workload{
 				*baseWorkloadBuilder.Clone().
@@ -835,6 +839,7 @@ func TestWlReconcile(t *testing.T) {
 			},
 		},
 		"remote wl is finished, the local workload and Job are marked completed (withoutJobManagedBy)": {
+			features:     map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
 			reconcileFor: "wl1",
 			managersWorkloads: []kueue.Workload{
 				*baseWorkloadBuilder.Clone().
diff --git a/pkg/controller/jobs/job/job_multikueue_adapter_test.go b/pkg/controller/jobs/job/job_multikueue_adapter_test.go
index 01e446758..2b2953648 100644
--- a/pkg/controller/jobs/job/job_multikueue_adapter_test.go
+++ b/pkg/controller/jobs/job/job_multikueue_adapter_test.go
@@ -394,6 +394,9 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 			},
 		},
 		"RemoteJobNotFound": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
+			},
 			args: args{
 				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
 					ManagedBy("parent").Obj()).Build(),
@@ -480,9 +483,12 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 					Obj(),
 			},
 		},
-		"RemoteJobFinished_Completed": {
+		"RemoteJobFinished_Completed_MultiKueueBatchJobWithManagedBy_Disabled": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
+			},
 			args: args{
-				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().Obj()).Build(),
+				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().Suspend(false).Obj()).Build(),
 				remoteClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
 					Condition(batchv1.JobCondition{
 						Type:   batchv1.JobComplete,
@@ -493,6 +499,7 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 			},
 			want: want{
 				localJob: newJob().
+					Suspend(false).
 					ResourceVersion("2").
 					Condition(
 						batchv1.JobCondition{
@@ -502,9 +509,38 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 					Obj(),
 			},
 		},
-		"RemoteJobFinished_Failed": {
+		"RemoteJobFinished_Completed_MultiKueueBatchJobWithManagedBy_Enabled": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: true},
+			},
 			args: args{
-				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().Obj()).Build(),
+				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().Suspend(false).Obj()).Build(),
+				remoteClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					Condition(batchv1.JobCondition{
+						Type:   batchv1.JobComplete,
+						Status: corev1.ConditionTrue,
+					}).
+					Obj()).Build(),
+				key: client.ObjectKeyFromObject(newJob().Obj()),
+			},
+			want: want{
+				localJob: newJob().
+					Condition(batchv1.JobCondition{
+						Type:   batchv1.JobComplete,
+						Status: corev1.ConditionTrue,
+					}).
+					Suspend(false).
+					ResourceVersion("2").
+					Obj(),
+			},
+		},
+		"RemoteJobFinished_Failed_MultiKueueBatchJobManagedBy_Disabled": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: false},
+			},
+
+			args: args{
+				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().Suspend(false).Obj()).Build(),
 				remoteClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
 					Condition(batchv1.JobCondition{
 						Type:   batchv1.JobFailed,
@@ -515,6 +551,7 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 			},
 			want: want{
 				localJob: newJob().
+					Suspend(false).
 					ResourceVersion("2").
 					Condition(
 						batchv1.JobCondition{
@@ -524,6 +561,33 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 					Obj(),
 			},
 		},
+		"RemoteJobFinished_Failed_MultiKueueBatchJobManagedBy_Enabled": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.MultiKueueBatchJobWithManagedBy: true},
+			},
+
+			args: args{
+				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().Suspend(false).Obj()).Build(),
+				remoteClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					Suspend(false).
+					Condition(batchv1.JobCondition{
+						Type:   batchv1.JobFailed,
+						Status: corev1.ConditionTrue,
+					}).
+					Obj()).Build(),
+				key: client.ObjectKeyFromObject(newJob().Obj()),
+			},
+			want: want{
+				localJob: newJob().
+					Suspend(false).
+					Condition(batchv1.JobCondition{
+						Type:   batchv1.JobFailed,
+						Status: corev1.ConditionTrue,
+					}).
+					ResourceVersion("2").
+					Obj(),
+			},
+		},
 		"ElasticJob_RemoteInSync": {
 			fields: fields{
 				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true},
@@ -584,9 +648,9 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 					Obj(),
 			},
 		},
-		"ElasticJob_WorkloadNameOnlyChange_EdgeCase": {
+		"ElasticJob_WorkloadNameOnlyChange_EdgeCase_MultiKueueBatchJobWithManagedBy_Disabled": {
 			fields: fields{
-				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true},
+				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true, features.MultiKueueBatchJobWithManagedBy: false},
 			},
 			args: args{
 				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
@@ -613,9 +677,37 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 					Obj(),
 			},
 		},
-		"ElasticJob_RemoteOutOfSync": {
+		"ElasticJob_WorkloadNameOnlyChange_EdgeCase_MultiKueueBatchJobWithManagedBy_Enabled": {
 			fields: fields{
-				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true},
+				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true, features.MultiKueueBatchJobWithManagedBy: true},
+			},
+			args: args{
+				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Condition(runningJobCondition).
+					Obj()).Build(),
+				remoteClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Condition(runningJobCondition).
+					Obj()).Build(),
+				key:          client.ObjectKeyFromObject(newJob().Obj()),
+				workloadName: "test-workload-new",
+			},
+			want: want{
+				localJob: newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Condition(runningJobCondition).
+					Obj(),
+				remoteJob: newJob().
+					ResourceVersion("1").
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Condition(runningJobCondition).
+					Obj(),
+			},
+		},
+		"ElasticJob_RemoteOutOfSync_MultiKueueBatchJobWithManagedBy_Disabled": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true, features.MultiKueueBatchJobWithManagedBy: false},
 			},
 			args: args{
 				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
@@ -646,9 +738,42 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 					Obj(),
 			},
 		},
-		"ElasticJob_RemoteOutOfSync_PatchFailure": {
+		"ElasticJob_RemoteOutOfSync_MultiKueueBatchJobWithManagedBy_Enabled": {
 			fields: fields{
-				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true},
+				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true, features.MultiKueueBatchJobWithManagedBy: true},
+			},
+			args: args{
+				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Parallelism(22).
+					Condition(runningJobCondition).
+					Obj()).Build(),
+				remoteClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Label(constants.PrebuiltWorkloadLabel, "test-workload").
+					Condition(runningJobCondition).
+					Obj()).Build(),
+				key:          client.ObjectKeyFromObject(newJob().Obj()),
+				workloadName: jobframework.GetWorkloadNameForOwnerWithGVKAndGeneration("test", "", gvk, 0),
+			},
+			want: want{
+				localJob: newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Parallelism(22).
+					Condition(runningJobCondition).
+					Obj(),
+				remoteJob: newJob().
+					ResourceVersion("1").
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Label(constants.PrebuiltWorkloadLabel, "test-workload").
+					Parallelism(1).
+					Condition(runningJobCondition).
+					Obj(),
+			},
+		},
+		"ElasticJob_RemoteOutOfSync_PatchFailure_MultiKueueBatchJobManagedBy_Disabled": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true, features.MultiKueueBatchJobWithManagedBy: false},
 			},
 			args: args{
 				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
@@ -683,6 +808,42 @@ func Test_multiKueueAdapter_SyncJob(t *testing.T) {
 					Obj(),
 			},
 		},
+		"ElasticJob_RemoteOutOfSync_PatchFailure_MultiKueueBatchJobManagedBy_Enabled": {
+			fields: fields{
+				features: map[featuregate.Feature]bool{features.ElasticJobsViaWorkloadSlices: true, features.MultiKueueBatchJobWithManagedBy: true},
+			},
+			args: args{
+				localClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Parallelism(22).
+					Condition(runningJobCondition).
+					Obj()).Build(),
+				remoteClient: fake.NewClientBuilder().WithScheme(schema).WithObjects(newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Label(constants.PrebuiltWorkloadLabel, "test-workload").
+					Condition(runningJobCondition).
+					Obj()).
+					WithInterceptorFuncs(interceptor.Funcs{
+						Patch: func(ctx context.Context, client client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
+							return errors.New("test-patch-error")
+						},
+					}).Build(),
+				key:          client.ObjectKeyFromObject(newJob().Obj()),
+				workloadName: jobframework.GetWorkloadNameForOwnerWithGVKAndGeneration("test", "", gvk, 0),
+			},
+			want: want{
+				localJob: newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Parallelism(22).
+					Condition(runningJobCondition).
+					Obj(),
+				remoteJob: newJob().
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Label(constants.PrebuiltWorkloadLabel, "test-workload").
+					Condition(runningJobCondition).
+					Obj(),
+			},
+		},
 	}
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
diff --git a/test/integration/multikueue/enabled_integration_test.go b/test/integration/multikueue/enabled_integration_test.go
index 1a94471fa..2678d6b2a 100644
--- a/test/integration/multikueue/enabled_integration_test.go
+++ b/test/integration/multikueue/enabled_integration_test.go
@@ -168,7 +168,7 @@ var _ = ginkgo.Describe("MultiKueue when not all integrations are enabled", gink
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
 				acs := admissioncheck.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAC.Name))
 				g.Expect(acs).NotTo(gomega.BeNil())
-				g.Expect(acs.State).To(gomega.Equal(kueue.CheckStatePending))
+				g.Expect(acs.State).To(gomega.Equal(kueue.CheckStateReady))
 				g.Expect(acs.Message).To(gomega.Equal(`The workload got reservation on "worker1"`))
 				ok, err := utiltesting.HasEventAppeared(managerTestCluster.ctx, managerTestCluster.client, corev1.Event{
 					Reason:  "MultiKueue",
diff --git a/test/integration/multikueue/jobs_test.go b/test/integration/multikueue/jobs_test.go
index 35a6891b8..cf1923732 100644
--- a/test/integration/multikueue/jobs_test.go
+++ b/test/integration/multikueue/jobs_test.go
@@ -245,7 +245,7 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
 				acs := admissioncheck.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAC.Name))
 				g.Expect(acs).NotTo(gomega.BeNil())
-				g.Expect(acs.State).To(gomega.Equal(kueue.CheckStatePending))
+				g.Expect(acs.State).To(gomega.Equal(kueue.CheckStateReady))
 				g.Expect(acs.Message).To(gomega.Equal(`The workload got reservation on "worker1"`))
 				ok, err := utiltesting.HasEventAppeared(managerTestCluster.ctx, managerTestCluster.client, corev1.Event{
 					Reason:  "MultiKueue",
@@ -1694,7 +1694,7 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 			localWorkload := getWorkload(g, manager.ctx, manager.client, workloadKey)
 			acs := admissioncheck.FindAdmissionCheck(localWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAC.Name))
 			g.Expect(acs).NotTo(gomega.BeNil())
-			g.Expect(acs.State).To(gomega.Equal(kueue.CheckStatePending))
+			g.Expect(acs.State).To(gomega.Equal(kueue.CheckStateReady))
 			g.Expect(acs.Message).To(gomega.Equal(`The workload got reservation on "worker1"`))
 			ok, err := utiltesting.HasEventAppeared(manager.ctx, manager.client, corev1.Event{
 				Reason:  "MultiKueue",
@@ -1724,9 +1724,9 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 			gomega.Expect(list.Items).To(gomega.BeEmpty())
 		})
 
-		ginkgo.By("observe: job is still suspended in the manager cluster", func() {
+		ginkgo.By("observe: job is no longer suspended in the manager cluster", func() {
 			getJob(manager.ctx, manager.client, job)
-			gomega.Expect(job.Spec.Suspend).To(gomega.BeEquivalentTo(ptr.To(true)))
+			gomega.Expect(job.Spec.Suspend).To(gomega.BeEquivalentTo(ptr.To(false)))
 		})
 
 		/*
@@ -1809,7 +1809,7 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 			workload := getWorkload(g, manager.ctx, manager.client, newWorkloadKey)
 			acs := admissioncheck.FindAdmissionCheck(workload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAC.Name))
 			g.Expect(acs).NotTo(gomega.BeNil())
-			g.Expect(acs.State).To(gomega.Equal(kueue.CheckStatePending))
+			g.Expect(acs.State).To(gomega.Equal(kueue.CheckStateReady))
 			g.Expect(acs.Message).To(gomega.Equal(`The workload got reservation on "worker1"`))
 			ok, err := utiltesting.HasEventAppeared(manager.ctx, manager.client, corev1.Event{
 				Reason:  "MultiKueue",
@@ -1825,7 +1825,7 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 				remoteJob := job.DeepCopy()
 				getJob(worker1.ctx, worker1.client, remoteJob)
 				g.Expect(remoteJob.Spec.Suspend).To(gomega.BeEquivalentTo(ptr.To(false)))
-				g.Expect(remoteJob.Spec.Parallelism).To(gomega.BeEquivalentTo(ptr.To(int32(2))))
+				g.Expect(remoteJob.Spec.Parallelism).To(gomega.BeEquivalentTo(ptr.To(int32(1))))
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})