Repository: kubernetes-sigs/kueue

Issue:
Support ElasticJobsViaWorkloadSlices for RayClusters
**What would you like to be added**:

I would like to add the support for ElasticJobs for RayClusters.

**Why is this needed**:

We have users requesting that feature as RayClusters are often long-living and may be handling varying load over time.


Discussion:
@mimowo: cc @ichekrygin @mwysokin @tenzen-y @tenzen-y: /assign @VassilisVassiliadis

@VassilisVassiliadis This is a tracking issue.@k8s-ci-robot: @tenzen-y: GitHub didn't allow me to assign the following users: VassilisVassiliadis.

Note that only [kubernetes-sigs members](https://github.com/orgs/kubernetes-sigs/people) with read permissions, repo collaborators and people who have commented on this issue/PR can be assigned. Additionally, issues/PRs can only have 10 assignees at the same time.
For more information please see [the contributor guide](https://git.k8s.io/community/contributors/guide/first-contribution.md#issue-assignment-in-github)

<details>

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

>/assign @VassilisVassiliadis
>
>@VassilisVassiliadis This is a tracking issue.


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>@tenzen-y: @VassilisVassiliadis Could you leave a comment with `/assign @VassilisVassiliadis`?@VassilisVassiliadis: /assign @VassilisVassiliadis@VassilisVassiliadis: I didn't spot this issue when I opened my PR the other day, sorry about that. I'm going to link it to this issue now.@tenzen-y: > I didn't spot this issue when I opened my PR the other day, sorry about that. I'm going to link it to this issue now.

Don't say sorry, thank you for your work!

Solution patch:
diff --git a/keps/77-dynamically-sized-jobs/README.md b/keps/77-dynamically-sized-jobs/README.md
index 592dc5992..ea018051e 100644
--- a/keps/77-dynamically-sized-jobs/README.md
+++ b/keps/77-dynamically-sized-jobs/README.md
@@ -30,13 +30,15 @@
     - [Scale Down](#scale-down)
     - [Scale Up](#scale-up)
   - [Phase 2 – RayCluster WorkloadSlice Support in Single-Cluster Configuration](#phase-2--raycluster-workloadslice-support-in-single-cluster-configuration)
+    - [Scale Down](#scale-down-1)
+    - [Scale Up](#scale-up-1)
   - [Phase 3 – Enabling Workload Slicing for batch/v1.Job in Multi-Cluster Configuration](#phase-3--enabling-workload-slicing-for-batchv1job-in-multi-cluster-configuration)
 - [Additional Details](#additional-details)
   - [Test Plan](#test-plan)
     - [Unit Tests](#unit-tests)
     - [Integration tests](#integration-tests)
-    - [Scale-down](#scale-down-1)
-    - [Scale-Down](#scale-down-2)
+    - [Scale-down](#scale-down-2)
+    - [Scale-Down](#scale-down-3)
     - [Resource Flavor Handling](#resource-flavor-handling)
   - [Graduation Criteria](#graduation-criteria)
   - [PodScheduling Readiness and ResourceQuota](#podscheduling-readiness-and-resourcequota)
@@ -326,6 +328,29 @@ Scaling up and down for `batch/v1.Job` will be the initial phase of the MVP, as
 ### Phase 2 – RayCluster WorkloadSlice Support in Single-Cluster Configuration
 This phase mirrors the steps from Phase 1 but is adapted specifically for RayCluster workloads, taking into account their autoscaling behavior and internal lifecycle management.
 
+
+#### Scale Down
+
+1. Create a `RayCluster` with workload-slice enablement, consisting of a head node and a worker group with 2 replicas. Both the head node and each replica in the worker group request 1 CPU core for a total of 3 CPU cores.
+2. Verify that a corresponding `Workload` is created, containing 2 `PodSets`, and that it is admitted. You should observe 3 pods in the `Running` state: one for the head node and two for the two worker group replicas.
+3. Update the job by reducing the number of replicas in the worker group from 2 to 1.
+4. Confirm that the existing `Workload` is updated accordingly:
+   - `WorkloadSpec.PodSets[0].Count` (for the head node) remains equal to 1.
+   - `WorkloadSpec.PodSets[1].Count` (for the worker group) is equal to 1 (down from 2).
+5. Observe that the number of running pods is reduced, while the remaining pods continue to run uninterrupted.
+6. Also observe that the ClusterQueue reflects the 2 CPUs that the RayCluster uses.
+
+#### Scale Up
+
+1. Create a `RayCluster` with workload-slice enablement, consisting of a head node and a worker group with 1 replica. Both the head node and each replica in the worker group request 1 CPU core.
+2. Verify that a corresponding `Workload` is created and admitted, and that there are 2 pods belonging to the `RayCluster` in the Running state: one for the head node and one for the worker group replica.
+3. Update the job by increasing the number of replicas in the worker group from 1 to 2.
+4. Confirm that a new `Workload` slice is created and admitted, reflecting the increased pod count, while the previous slice is marked as `Finished`. The new `Workload` reflects the current state of the `RayCluster`:
+   - `WorkloadSpec.PodSets[0].Count` (for the head node) is equal to 1.
+   - `WorkloadSpec.PodSets[1].Count` (for the worker group) is equal to 2.
+5. Observe that the total number of running pods increases to 3 to match the updated replica count, with the original pods continuing to run without disruption.
+6. Also observe that the ClusterQueue reflects the 3 CPUs that the RayCluster uses.
+
 ### Phase 3 – Enabling Workload Slicing for batch/v1.Job in Multi-Cluster Configuration
 In this phase, Workload Slicing support for batch/v1.Job will be extended to multi-cluster environments using Kueue’s MultiKueue architecture. 
 When a job is scaled in the management cluster, its corresponding WorkloadSlice will be propagated to the appropriate workload cluster(s), respecting existing cluster assignment and resource flavor constraints. 
diff --git a/pkg/controller/jobs/raycluster/raycluster_webhook.go b/pkg/controller/jobs/raycluster/raycluster_webhook.go
index 02b168732..071478ed8 100644
--- a/pkg/controller/jobs/raycluster/raycluster_webhook.go
+++ b/pkg/controller/jobs/raycluster/raycluster_webhook.go
@@ -19,8 +19,10 @@ package raycluster
 import (
 	"context"
 	"fmt"
+	"slices"
 
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
@@ -35,7 +37,9 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework/webhook"
 	"sigs.k8s.io/kueue/pkg/features"
+	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 	"sigs.k8s.io/kueue/pkg/util/podset"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 )
 
 var (
@@ -87,6 +91,20 @@ func (w *RayClusterWebhook) Default(ctx context.Context, obj runtime.Object) err
 		return err
 	}
 	jobframework.ApplyDefaultForManagedBy(job, w.queues, w.cache, log)
+
+	rjob := obj.(*rayv1.RayCluster)
+
+	if isAnElasticJob(rjob) {
+		// Ensure that the PodSchedulingGate is present in the RayCluster's pod Templates for its Head and all its Workers
+		utilpod.GateTemplate(&job.Spec.HeadGroupSpec.Template, kueuebeta.ElasticJobSchedulingGate)
+
+		for index := range job.Spec.WorkerGroupSpecs {
+			wgs := &job.Spec.WorkerGroupSpecs[index]
+
+			utilpod.GateTemplate(&wgs.Template, kueuebeta.ElasticJobSchedulingGate)
+		}
+	}
+
 	return nil
 }
 
@@ -106,6 +124,11 @@ func (w *RayClusterWebhook) ValidateCreate(ctx context.Context, obj runtime.Obje
 	return nil, validationErrs.ToAggregate()
 }
 
+// returns whether the RayCluster is an elastic job or not
+func isAnElasticJob(job *rayv1.RayCluster) bool {
+	return features.Enabled(features.ElasticJobsViaWorkloadSlices) && workloadslicing.Enabled(job.GetObjectMeta())
+}
+
 func (w *RayClusterWebhook) validateCreate(job *rayv1.RayCluster) (field.ErrorList, error) {
 	var allErrors field.ErrorList
 	kueueJob := (*RayCluster)(job)
@@ -114,10 +137,11 @@ func (w *RayClusterWebhook) validateCreate(job *rayv1.RayCluster) (field.ErrorLi
 		spec := &job.Spec
 		specPath := field.NewPath("spec")
 
-		// TODO revisit once Support dynamically sized (elastic) jobs #77 is implemented
-		// Should not use auto scaler. Once the resources are reserved by queue the cluster should do it's best to use them.
-		if ptr.Deref(spec.EnableInTreeAutoscaling, false) {
-			allErrors = append(allErrors, field.Invalid(specPath.Child("enableInTreeAutoscaling"), spec.EnableInTreeAutoscaling, "a kueue managed job should not use autoscaling"))
+		if isAnElasticJob(job) {
+			allErrors = append(allErrors, validateElasticJob(job)...)
+		} else if ptr.Deref(spec.EnableInTreeAutoscaling, false) {
+			// Should not use auto scaler. Once the resources are reserved by queue the cluster should do its best to use them.
+			allErrors = append(allErrors, field.Invalid(specPath.Child("enableInTreeAutoscaling"), spec.EnableInTreeAutoscaling, "a kueue managed job can use autoscaling only when the ElasticJobsViaWorkloadSlices feature gate is on and the job is an elastic job"))
 		}
 
 		// Should limit the worker count to 8 - 1 (max podSets num - cluster head)
@@ -145,6 +169,30 @@ func (w *RayClusterWebhook) validateCreate(job *rayv1.RayCluster) (field.ErrorLi
 	return allErrors, nil
 }
 
+func validateElasticJob(job *rayv1.RayCluster) field.ErrorList {
+	allErrors := field.ErrorList{}
+
+	specPath := field.NewPath("spec")
+
+	workloadSliceSchedulingGate := corev1.PodSchedulingGate{
+		Name: kueuebeta.ElasticJobSchedulingGate,
+	}
+
+	for index := range job.Spec.WorkerGroupSpecs {
+		wgs := &job.Spec.WorkerGroupSpecs[index]
+
+		if !slices.Contains(wgs.Template.Spec.SchedulingGates, workloadSliceSchedulingGate) {
+			allErrors = append(allErrors, field.Invalid(specPath.Child("workerGroupSpecs").Index(index).Child("template").Child("spec").Child("schedulingGates"), wgs.Template.Spec.SchedulingGates, "an elastic job must have the ElasticJobSchedulingGate"))
+		}
+	}
+
+	if !slices.Contains(job.Spec.HeadGroupSpec.Template.Spec.SchedulingGates, workloadSliceSchedulingGate) {
+		allErrors = append(allErrors, field.Invalid(specPath.Child("headGroupSpec").Child("template").Child("spec").Child("schedulingGates"), job.Spec.HeadGroupSpec.Template.Spec.SchedulingGates, "an elastic job must have the ElasticJobSchedulingGate"))
+	}
+
+	return allErrors
+}
+
 func (w *RayClusterWebhook) validateTopologyRequest(rayJob *RayCluster) (field.ErrorList, error) {
 	var allErrs field.ErrorList
 
diff --git a/pkg/controller/jobs/rayjob/rayjob_webhook.go b/pkg/controller/jobs/rayjob/rayjob_webhook.go
index d5cf7fdae..e9f3c0658 100644
--- a/pkg/controller/jobs/rayjob/rayjob_webhook.go
+++ b/pkg/controller/jobs/rayjob/rayjob_webhook.go
@@ -124,7 +124,7 @@ func (w *RayJobWebhook) validateCreate(job *rayv1.RayJob) (field.ErrorList, erro
 		clusterSpec := spec.RayClusterSpec
 		clusterSpecPath := specPath.Child("rayClusterSpec")
 
-		// Should not use auto scaler. Once the resources are reserved by queue the cluster should do it's best to use them.
+		// Should not use auto scaler. Once the resources are reserved by queue the cluster should do its best to use them.
 		if ptr.Deref(clusterSpec.EnableInTreeAutoscaling, false) {
 			allErrors = append(allErrors, field.Invalid(clusterSpecPath.Child("enableInTreeAutoscaling"), clusterSpec.EnableInTreeAutoscaling, "a kueue managed job should not use autoscaling"))
 		}
diff --git a/site/content/en/docs/concepts/elastic_workload.md b/site/content/en/docs/concepts/elastic_workload.md
index 9bffc0ff9..ab81ceaed 100644
--- a/site/content/en/docs/concepts/elastic_workload.md
+++ b/site/content/en/docs/concepts/elastic_workload.md
@@ -62,7 +62,9 @@ metadata:
 
 ## Limitations
 
-* Currently available only for `batch/v1.Job` workloads.
+* Currently available only for the following workloads: 
+   * `batch/v1.Job`
+   * `ray.io/v1.RayCluster`
 * Elastic workloads are not supported for jobs with partial admission enabled.
 
     * Attempting to scale jobs with partial admission enabled will result in an admission validation error similar to the following:
diff --git a/site/content/en/docs/tasks/run/rayclusters.md b/site/content/en/docs/tasks/run/rayclusters.md
index 5afed9d4a..7d2e846bb 100644
--- a/site/content/en/docs/tasks/run/rayclusters.md
+++ b/site/content/en/docs/tasks/run/rayclusters.md
@@ -65,7 +65,22 @@ Note that a RayCluster will hold resource quotas while it exists. For optimal re
 
 ### c. Limitations
 - Limited Worker Groups: Because a Kueue workload can have a maximum of 8 PodSets, the maximum number of `spec.workerGroupSpecs` is 7
-- In-Tree Autoscaling Disabled: Kueue manages resource allocation for the RayCluster; therefore, the cluster's internal autoscaling mechanisms need to be disabled
+- In-Tree Autoscaling Constraints: Autoscaling is only supported for [elastic](/docs/concepts/elastic_workload) RayCluster objects. To enable in-tree autoscaling:
+
+  1. Activate the `ElasticJobsViaWorkloadSlices` feature gate.
+  2. Annotate the RayCluster object with:
+
+     ```yaml
+     metadata:
+       annotations:
+         kueue.x-k8s.io/elastic-job: "true"
+     ```
+  3. Enable the Ray autoscaler of your RayCluster object by setting:
+
+     ```yaml
+     spec:
+       enableInTreeAutoscaling: true
+     ```
 
 ## Example RayCluster
 



Test patch:
diff --git a/pkg/controller/jobs/raycluster/raycluster_webhook_test.go b/pkg/controller/jobs/raycluster/raycluster_webhook_test.go
index 7db59b4a3..bc929b27b 100644
--- a/pkg/controller/jobs/raycluster/raycluster_webhook_test.go
+++ b/pkg/controller/jobs/raycluster/raycluster_webhook_test.go
@@ -149,7 +149,7 @@ func TestValidateCreate(t *testing.T) {
 				WithEnableAutoscaling(ptr.To(true)).
 				Obj(),
 			wantErr: field.ErrorList{
-				field.Invalid(field.NewPath("spec", "enableInTreeAutoscaling"), ptr.To(true), "a kueue managed job should not use autoscaling"),
+				field.Invalid(field.NewPath("spec", "enableInTreeAutoscaling"), ptr.To(true), "a kueue managed job can use autoscaling only when the ElasticJobsViaWorkloadSlices feature gate is on and the job is an elastic job"),
 			}.ToAggregate(),
 		},
 		"invalid managed - too many worker groups": {
diff --git a/pkg/util/testingjobs/raycluster/wrappers.go b/pkg/util/testingjobs/raycluster/wrappers.go
index 2edc0b679..5f576c2a6 100644
--- a/pkg/util/testingjobs/raycluster/wrappers.go
+++ b/pkg/util/testingjobs/raycluster/wrappers.go
@@ -133,6 +133,11 @@ func (j *ClusterWrapper) Queue(queue string) *ClusterWrapper {
 	return j
 }
 
+func (j *ClusterWrapper) SetAnnotation(key, content string) *ClusterWrapper {
+	j.Annotations[key] = content
+	return j
+}
+
 // Clone returns deep copy of the Job.
 func (j *ClusterWrapper) Clone() *ClusterWrapper {
 	return &ClusterWrapper{RayCluster: *j.DeepCopy()}
@@ -143,6 +148,11 @@ func (j *ClusterWrapper) WithEnableAutoscaling(value *bool) *ClusterWrapper {
 	return j
 }
 
+func (j *ClusterWrapper) ScaleFirstWorkerGroup(replicas int32) *ClusterWrapper {
+	j.Spec.WorkerGroupSpecs[0].Replicas = &replicas
+	return j
+}
+
 func (j *ClusterWrapper) WithWorkerGroups(workers ...rayv1.WorkerGroupSpec) *ClusterWrapper {
 	j.Spec.WorkerGroupSpecs = workers
 	return j
diff --git a/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go b/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go
index 9dfc7544f..ec72973f6 100644
--- a/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go
@@ -18,6 +18,7 @@ package raycluster
 
 import (
 	"fmt"
+	"strconv"
 
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
@@ -25,8 +26,10 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	schedulingv1 "k8s.io/api/scheduling/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -36,9 +39,12 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	workloadraycluster "sigs.k8s.io/kueue/pkg/controller/jobs/raycluster"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	testingraycluster "sigs.k8s.io/kueue/pkg/util/testingjobs/raycluster"
 	testingrayjob "sigs.k8s.io/kueue/pkg/util/testingjobs/rayjob"
+	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 	"sigs.k8s.io/kueue/test/util"
 
 	_ "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob" // to enable the framework
@@ -690,3 +696,254 @@ var _ = ginkgo.Describe("Job controller with preemption enabled", ginkgo.Ordered
 		}, util.Timeout, util.Interval).Should(gomega.Succeed())
 	})
 })
+
+var _ = ginkgo.Describe("RayCluster with elastic jobs via workload-slices support", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		ns             *corev1.Namespace
+		resourceFlavor *kueue.ResourceFlavor
+		clusterQueue   *kueue.ClusterQueue
+		localQueue     *kueue.LocalQueue
+	)
+	cpuNominalQuota := 5
+
+	ginkgo.BeforeAll(func() {
+		gomega.Expect(utilfeature.DefaultMutableFeatureGate.SetFromMap(map[string]bool{string(features.ElasticJobsViaWorkloadSlices): true})).Should(gomega.Succeed())
+		fwk.StartManager(ctx, cfg, managerAndSchedulerSetup())
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+
+	ginkgo.BeforeEach(func() {
+		ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "core-")
+
+		resourceFlavor = testing.MakeResourceFlavor("default").Obj()
+		util.MustCreate(ctx, k8sClient, resourceFlavor)
+
+		clusterQueue = testing.MakeClusterQueue("default").
+			ResourceGroup(*testing.MakeFlavorQuotas(resourceFlavor.Name).Resource(corev1.ResourceCPU, strconv.Itoa(cpuNominalQuota)).Obj()).
+			Preemption(kueue.ClusterQueuePreemption{
+				WithinClusterQueue: kueue.PreemptionPolicyLowerPriority,
+			}).
+			Obj()
+		util.MustCreate(ctx, k8sClient, clusterQueue)
+
+		localQueue = testing.MakeLocalQueue("default", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+		util.MustCreate(ctx, k8sClient, localQueue)
+	})
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, localQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, resourceFlavor, true)
+	})
+
+	ginkgo.It("Should support raycluster scale-down and scale-up", func() {
+		testRayCluster := testingraycluster.MakeCluster("foo", ns.Name).
+			SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+			Queue(string(kueue.LocalQueueName(localQueue.Name))).
+			Request(rayv1.HeadNode, corev1.ResourceCPU, "1").
+			RequestWorkerGroup(corev1.ResourceCPU, "1").
+			WithEnableAutoscaling(ptr.To(true)).
+			ScaleFirstWorkerGroup(2).
+			Obj()
+
+		var testRayClusterWorkload *kueue.Workload
+
+		ginkgo.By("creating a raycluster")
+		util.MustCreate(ctx, k8sClient, testRayCluster)
+
+		ginkgo.By("admitting the raycluster's workload")
+		gomega.Eventually(func(g gomega.Gomega) {
+			workloads := &kueue.WorkloadList{}
+			g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testRayCluster.Namespace))).Should(gomega.Succeed())
+			g.Expect(workloads.Items).Should(gomega.HaveLen(1))
+			testRayClusterWorkload = &workloads.Items[0]
+			g.Expect(testRayClusterWorkload.Spec.PodSets).Should(gomega.HaveLen(2))
+
+			g.Expect(testRayClusterWorkload.Spec.PodSets[0].Count).Should(gomega.BeEquivalentTo(int32(1)))
+			g.Expect(testRayClusterWorkload.Spec.PodSets[1].Count).Should(gomega.BeEquivalentTo(int32(2)))
+
+			util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testRayClusterWorkload)
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("the raycluster is unsuspended")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testRayCluster), testRayCluster)).Should(gomega.Succeed())
+			g.Expect(ptr.Deref(testRayCluster.Spec.Suspend, false)).Should(gomega.BeFalse())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("resource flavor utilization is correctly recorded")
+		gomega.Eventually(func(g gomega.Gomega) {
+			cq := &kueue.ClusterQueue{}
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), cq)).Should(gomega.Succeed())
+			g.Expect(len(cq.Status.FlavorsUsage)).Should(gomega.BeEquivalentTo(1))
+			g.Expect(len(cq.Status.FlavorsUsage[0].Resources)).Should(gomega.BeEquivalentTo(1))
+			g.Expect(cq.Status.FlavorsUsage[0].Resources[0].Total).Should(gomega.BeEquivalentTo(resource.MustParse("3")))
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("reducing the worker replicas to 1 to emulate scale-down operation")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testRayCluster), testRayCluster)).Should(gomega.Succeed())
+			testRayCluster.Spec.WorkerGroupSpecs[0].Replicas = ptr.To(int32(1))
+			g.Expect(k8sClient.Update(ctx, testRayCluster)).Should(gomega.Succeed())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("resource flavor utilization is correctly updated")
+		gomega.Eventually(func(g gomega.Gomega) {
+			cq := &kueue.ClusterQueue{}
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), cq)).Should(gomega.Succeed())
+			g.Expect(len(cq.Status.FlavorsUsage)).Should(gomega.BeEquivalentTo(1))
+			g.Expect(len(cq.Status.FlavorsUsage[0].Resources)).Should(gomega.BeEquivalentTo(1))
+			g.Expect(cq.Status.FlavorsUsage[0].Resources[0].Total).Should(gomega.BeEquivalentTo(resource.MustParse("2")))
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("assert the raycluster's workload is updated and still admitted")
+		gomega.Eventually(func(g gomega.Gomega) {
+			workloads := &kueue.WorkloadList{}
+			g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testRayCluster.Namespace))).Should(gomega.Succeed())
+			g.Expect(workloads.Items).Should(gomega.HaveLen(1))
+			util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testRayClusterWorkload)
+
+			g.Expect(workloads.Items[0].Spec.PodSets).Should(gomega.HaveLen(2))
+			g.Expect(workloads.Items[0].Spec.PodSets[0].Count).Should(gomega.Equal(int32(1)))
+			g.Expect(workloads.Items[0].Spec.PodSets[1].Count).Should(gomega.Equal(int32(1)))
+
+			g.Expect(workloads.Items[0].UID).Should(gomega.BeEquivalentTo(testRayClusterWorkload.UID))
+			testRayClusterWorkload = &workloads.Items[0]
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("increasing the RayCluster's worker replicas to 2 to emulate scale-up operation")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testRayCluster), testRayCluster)).Should(gomega.Succeed())
+			testRayCluster.Spec.WorkerGroupSpecs[0].Replicas = ptr.To(int32(2))
+			g.Expect(k8sClient.Update(ctx, testRayCluster)).Should(gomega.Succeed())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("resource flavor utilization is correctly updated")
+		gomega.Eventually(func(g gomega.Gomega) {
+			cq := &kueue.ClusterQueue{}
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), cq)).Should(gomega.Succeed())
+			g.Expect(len(cq.Status.FlavorsUsage)).Should(gomega.BeEquivalentTo(1))
+			g.Expect(len(cq.Status.FlavorsUsage[0].Resources)).Should(gomega.BeEquivalentTo(1))
+			// 1 core for the head node, and 1 for each of the 2 workers
+			g.Expect(cq.Status.FlavorsUsage[0].Resources[0].Total).Should(gomega.BeEquivalentTo(resource.MustParse("3")))
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("old workload is finished and new workload is admitted")
+		gomega.Eventually(func(g gomega.Gomega) {
+			workloads := &kueue.WorkloadList{}
+			g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testRayCluster.Namespace))).Should(gomega.Succeed())
+			g.Expect(workloads.Items).Should(gomega.HaveLen(2))
+
+			nameBeforeScaleUp := testRayClusterWorkload.Name
+
+			for i := range workloads.Items {
+				if workload.IsFinished(&workloads.Items[i]) {
+					g.Expect(workloads.Items[i].Name).Should(gomega.Equal(nameBeforeScaleUp))
+
+					// The original workload had 1 head node, and a single worker group with a single worker
+					g.Expect(workloads.Items[i].Spec.PodSets).Should(gomega.HaveLen(2))
+					g.Expect(workloads.Items[i].Spec.PodSets[0].Count).Should(gomega.Equal(int32(1)))
+					g.Expect(workloads.Items[i].Spec.PodSets[1].Count).Should(gomega.Equal(int32(1)))
+				} else {
+					g.Expect(workloads.Items[i].Name).ShouldNot(gomega.Equal(testRayClusterWorkload.Name))
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testRayClusterWorkload)
+					testRayClusterWorkload = &workloads.Items[i]
+				}
+			}
+
+			// We should have both the finished and the non-finished but admitted workload
+			g.Expect(testRayClusterWorkload.Name).ShouldNot(gomega.Equal(nameBeforeScaleUp))
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("the new workload has the correct podSets count")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(workload.IsAdmitted(testRayClusterWorkload)).Should(gomega.BeTrue())
+			g.Expect(workload.IsFinished(testRayClusterWorkload)).Should(gomega.BeFalse())
+
+			g.Expect(testRayClusterWorkload.Spec.PodSets).Should(gomega.HaveLen(2))
+			g.Expect(testRayClusterWorkload.Spec.PodSets[0].Count).Should(gomega.Equal(int32(1)))
+			g.Expect(testRayClusterWorkload.Spec.PodSets[1].Count).Should(gomega.Equal(int32(2)))
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+	})
+
+	ginkgo.It("Should support scheduling pending workload after freeing capacity on scale-down", func() {
+		var (
+			testRayClusterAWorkload *kueue.Workload
+			testRayClusterBWorkload *kueue.Workload
+		)
+
+		testRayClusterA := testingraycluster.MakeCluster("raycluster-a", ns.Name).
+			SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+			Queue(string(kueue.LocalQueueName(localQueue.Name))).
+			Request(rayv1.HeadNode, corev1.ResourceCPU, "1").
+			RequestWorkerGroup(corev1.ResourceCPU, "1").
+			WithEnableAutoscaling(ptr.To(true)).
+			ScaleFirstWorkerGroup(2).
+			Obj()
+
+		ginkgo.By("creating a raycluster-a")
+		util.MustCreate(ctx, k8sClient, testRayClusterA)
+
+		ginkgo.By("admitting the raycluster-a's workload")
+		gomega.Eventually(func(g gomega.Gomega) {
+			workloads := &kueue.WorkloadList{}
+			g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testRayClusterA.Namespace))).Should(gomega.Succeed())
+			g.Expect(workloads.Items).Should(gomega.HaveLen(1))
+			testRayClusterAWorkload = &workloads.Items[0]
+			util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testRayClusterAWorkload)
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("the raycluster-a is unsuspended")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testRayClusterA), testRayClusterA)).Should(gomega.Succeed())
+			g.Expect(ptr.Deref(testRayClusterA.Spec.Suspend, false)).Should(gomega.BeFalse())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		testRayClusterB := testingraycluster.MakeCluster("raycluster-b", ns.Name).
+			SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+			Queue(string(kueue.LocalQueueName(localQueue.Name))).
+			Request(rayv1.HeadNode, corev1.ResourceCPU, "1").
+			RequestWorkerGroup(corev1.ResourceCPU, "1").
+			WithEnableAutoscaling(ptr.To(true)).
+			ScaleFirstWorkerGroup(2).
+			Obj()
+
+		ginkgo.By("creating a raycluster-b")
+		util.MustCreate(ctx, k8sClient, testRayClusterB)
+
+		ginkgo.By("the testRayClusterB remains suspended")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testRayClusterB), testRayClusterB)).Should(gomega.Succeed())
+			g.Expect(ptr.Deref(testRayClusterB.Spec.Suspend, false)).Should(gomega.BeTrue())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("raycluster-b's workload remains pending with unreserved quota")
+		gomega.Eventually(func(g gomega.Gomega) {
+			workloads := &kueue.WorkloadList{}
+			g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testRayClusterA.Namespace), client.MatchingLabels{constants.JobUIDLabel: string(testRayClusterB.UID)})).Should(gomega.Succeed())
+			g.Expect(workloads.Items).Should(gomega.HaveLen(1))
+			testRayClusterBWorkload = &workloads.Items[0]
+			util.ExpectWorkloadsToBePending(ctx, k8sClient, testRayClusterBWorkload)
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("scale-down raycluster-a to make room for raycluster-b")
+		gomega.Eventually(func(g gomega.Gomega) {
+			testRayClusterA.Spec.WorkerGroupSpecs[0].Replicas = ptr.To(int32(1))
+			g.Expect(k8sClient.Update(ctx, testRayClusterA)).Should(gomega.Succeed())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("admitting the raycluster-b workload")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testRayClusterBWorkload), testRayClusterBWorkload)).Should(gomega.Succeed())
+			util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testRayClusterBWorkload)
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+		ginkgo.By("the raycluster-b is unsuspended")
+		gomega.Eventually(func(g gomega.Gomega) {
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testRayClusterB), testRayClusterB)).Should(gomega.Succeed())
+			g.Expect(ptr.Deref(testRayClusterB.Spec.Suspend, false)).Should(gomega.BeFalse())
+		}, util.Timeout, util.Interval).Should(gomega.Succeed())
+	})
+})
diff --git a/test/util/util.go b/test/util/util.go
index 466b5afaf..d3eec9060 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -937,6 +937,14 @@ func ExpectJobUnsuspendedWithNodeSelectors(ctx context.Context, c client.Client,
 	}, Timeout, Interval).Should(gomega.Succeed())
 }
 
+func ExpectRayClusterUnsuspended(ctx context.Context, c client.Client, key types.NamespacedName) {
+	rayCluster := &rayv1.RayCluster{}
+	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
+		g.Expect(c.Get(ctx, key, rayCluster)).To(gomega.Succeed())
+		g.Expect(rayCluster.Spec.Suspend).Should(gomega.Equal(ptr.To(false)))
+	}, Timeout, Interval).Should(gomega.Succeed())
+}
+
 func CreateNodesWithStatus(ctx context.Context, c client.Client, nodes []corev1.Node) {
 	for _, node := range nodes {
 		// 1. Create a node