Repository: kubernetes-sigs/kueue

Issue:
Specify the platform for Ray/Raymini images using env. variable
<!-- Please only use this template for submitting clean up requests -->

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

Specify the images for Ray/Raymini images using the env. variable. We probably could introduce RAY_PLATFORMS or RAY_PLATFORM variable.

**Why is this needed**:

To avoid reading the architecture from the os, and allow choosing different architectures. This was discussed in https://github.com/kubernetes-sigs/kueue/pull/4917#discussion_r2060768176

Discussion:
@mimowo: cc @tenzen-y @mszadkow @tenzen-y: We are using the `PLATFORMS` env vars for image buiding during E2E testings:

https://github.com/kubernetes-sigs/kueue/blob/3d4725b8553082570b86ad7b81120570c26ea5fa/Makefile#L242-L244

Do we want to introduce a separate env var? @mszadkow Do you have any concerns about reusing the `PLATFORMS`?
@mimowo: If reusing works, I'm ok with that, even better@mszadkow: I will definitively check possibility to reuse this.@tenzen-y: > I will definitively check possibility to reuse this.

thank you!@mszadkow: /assign


Solution patch:
diff --git a/Makefile b/Makefile
index 84eb06903..b80fefaf8 100644
--- a/Makefile
+++ b/Makefile
@@ -406,7 +406,7 @@ generate-kueuectl-docs: kueuectl-docs
 .PHONY: ray-project-mini-image-build
 ray-project-mini-image-build:
 	$(IMAGE_BUILD_CMD) \
-		-t $(IMAGE_REGISTRY)/ray-project-mini:$(RAYMINI_VERSION)$(RAY_ARCHITECTURE) \
+		-t $(IMAGE_REGISTRY)/ray-project-mini:$(RAYMINI_VERSION) \
 		--platform=$(PLATFORMS) \
 		--build-arg RAY_VERSION=$(RAY_VERSION) \
 		$(PUSH) \
@@ -414,9 +414,6 @@ ray-project-mini-image-build:
 
 # The step is required for local e2e test run
 .PHONY: kind-ray-project-mini-image-build
-kind-ray-project-mini-image-build:
-	@if [ "$(shell uname -m)" = "arm64" ]; then \
-		$(MAKE) PLATFORMS=linux/arm64  RAY_ARCHITECTURE=-aarch64 PUSH=--load  ray-project-mini-image-build; \
-	else \
-		$(MAKE) PLATFORMS=linux/amd64 PUSH=--load ray-project-mini-image-build; \
-	fi
+kind-ray-project-mini-image-build: PLATFORMS=linux/amd64
+kind-ray-project-mini-image-build: PUSH=--load
+kind-ray-project-mini-image-build: ray-project-mini-image-build
\ No newline at end of file
diff --git a/hack/e2e-common.sh b/hack/e2e-common.sh
index 32a4edbc8..e872d2833 100644
--- a/hack/e2e-common.sh
+++ b/hack/e2e-common.sh
@@ -246,27 +246,15 @@ function restore_managers_image {
 
 function determine_kuberay_ray_image {
     local RAY_IMAGE=rayproject/ray:${RAY_VERSION}
-    local RAY_IMAGE_ARM=rayproject/ray:${RAY_VERSION}-aarch64
     local RAYMINI_IMAGE=us-central1-docker.pkg.dev/k8s-staging-images/kueue/ray-project-mini:${RAYMINI_VERSION}
-    local RAYMINI_IMAGE_ARM=us-central1-docker.pkg.dev/k8s-staging-images/kueue/ray-project-mini:${RAYMINI_VERSION}-aarch64
 
     # Extra e2e images required for Kuberay
-    local unamestr=""
     local ray_image_to_use=""
 
-    unamestr=$(uname)
     if [[ "${USE_RAY_FOR_TESTS:-}" == "ray" ]]; then
-        if [[ "$unamestr" == "Linux" ]]; then
-            ray_image_to_use="${RAY_IMAGE}"
-        elif [[ "$unamestr" == "Darwin" ]]; then
-            ray_image_to_use="${RAY_IMAGE_ARM}"
-        fi
+        ray_image_to_use="${RAY_IMAGE}"
     else
-        if [[ "$unamestr" == "Linux" ]]; then
-            ray_image_to_use="${RAYMINI_IMAGE}"
-        elif [[ "$unamestr" == "Darwin" ]]; then
-            ray_image_to_use="${RAYMINI_IMAGE_ARM}"
-        fi
+        ray_image_to_use="${RAYMINI_IMAGE}"
     fi
 
     if [[ -z "$ray_image_to_use" ]]; then
diff --git a/pkg/cache/clusterqueue.go b/pkg/cache/clusterqueue.go
index 925051de7..abce8824a 100644
--- a/pkg/cache/clusterqueue.go
+++ b/pkg/cache/clusterqueue.go
@@ -92,7 +92,7 @@ func (c *clusterQueue) GetName() kueue.ClusterQueueReference {
 	return c.Name
 }
 
-// implements hierarchicalResourceNode interface.
+// implement flatResourceNode/hierarchicalResourceNode interfaces
 
 func (c *clusterQueue) getResourceNode() resourceNode {
 	return c.resourceNode
diff --git a/pkg/cache/clusterqueue_snapshot.go b/pkg/cache/clusterqueue_snapshot.go
index 1e0d514ee..d2a610e95 100644
--- a/pkg/cache/clusterqueue_snapshot.go
+++ b/pkg/cache/clusterqueue_snapshot.go
@@ -186,7 +186,7 @@ func (c *ClusterQueueSnapshot) fairWeight() *resource.Quantity {
 	return &c.FairWeight
 }
 
-// The methods below implement hierarchicalResourceNode interface.
+// implement flatResourceNode/hierarchicalResourceNode interfaces
 
 func (c *ClusterQueueSnapshot) getResourceNode() resourceNode {
 	return c.ResourceNode
diff --git a/pkg/cache/cohort.go b/pkg/cache/cohort.go
index ec04a5112..931a86927 100644
--- a/pkg/cache/cohort.go
+++ b/pkg/cache/cohort.go
@@ -66,7 +66,7 @@ func (c *cohort) getRootUnsafe() *cohort {
 	return c.Parent().getRootUnsafe()
 }
 
-// implements hierarchicalResourceNode interface.
+// implement flatResourceNode/hierarchicalResourceNode interfaces
 
 func (c *cohort) getResourceNode() resourceNode {
 	return c.resourceNode
diff --git a/pkg/cache/cohort_snapshot.go b/pkg/cache/cohort_snapshot.go
index bf74056e6..6a5a8bc4c 100644
--- a/pkg/cache/cohort_snapshot.go
+++ b/pkg/cache/cohort_snapshot.go
@@ -74,7 +74,7 @@ func (c *CohortSnapshot) DominantResourceShare() int {
 	return share
 }
 
-// The methods below implement hierarchicalResourceNode interface.
+// implement flatResourceNode/hierarchicalResourceNode interfaces
 
 func (c *CohortSnapshot) getResourceNode() resourceNode {
 	return c.ResourceNode
diff --git a/pkg/cache/resource_node.go b/pkg/cache/resource_node.go
index 9d4062798..4c0ea3926 100644
--- a/pkg/cache/resource_node.go
+++ b/pkg/cache/resource_node.go
@@ -69,20 +69,25 @@ func (r resourceNode) guaranteedQuota(fr resources.FlavorResource) int64 {
 	return 0
 }
 
-// hierarchicalResourceNode abstracts over ClusterQueues and Cohorts,
-// by providing access to the contained ResourceNode, with the ability
-// to navigate to the parent node.
+// hierarchicalResourceNode extends flatResourceNode
+// with the ability to navigate to the parent node.
 type hierarchicalResourceNode interface {
-	getResourceNode() resourceNode
+	flatResourceNode
 
 	HasParent() bool
 	parentHRN() hierarchicalResourceNode
 }
 
+// flatResourceNode abstracts over ClusterQueues and Cohorts,
+// by providing access to the contained ResourceNode.
+type flatResourceNode interface {
+	getResourceNode() resourceNode
+}
+
 // LocalAvailable returns, for a given node and resource flavor,
 // how much guaranteed quota in this flavor exceeds usage.
 // This quota is available at this node but is not visible at its parent.
-func LocalAvailable(node hierarchicalResourceNode, fr resources.FlavorResource) int64 {
+func LocalAvailable(node flatResourceNode, fr resources.FlavorResource) int64 {
 	return max(0, node.getResourceNode().guaranteedQuota(fr)-node.getResourceNode().Usage[fr])
 }
 
@@ -191,7 +196,7 @@ func updateCohortResourceNode(cohort *cohort) {
 	}
 }
 
-func accumulateFromChild(parent *cohort, child hierarchicalResourceNode) {
+func accumulateFromChild(parent *cohort, child flatResourceNode) {
 	for fr, childQuota := range child.getResourceNode().SubtreeQuota {
 		parent.resourceNode.SubtreeQuota[fr] += childQuota - child.getResourceNode().guaranteedQuota(fr)
 	}
@@ -204,7 +209,7 @@ func accumulateFromChild(parent *cohort, child hierarchicalResourceNode) {
 // and the amount of resources that exceed the guaranteed
 // quota on this node. It is assumed that subsequent call
 // to this function will be on nodes parent with remainingRequests
-func QuantitiesFitInQuota(node hierarchicalResourceNode, requests resources.FlavorResourceQuantities) (bool, resources.FlavorResourceQuantities) {
+func QuantitiesFitInQuota(node flatResourceNode, requests resources.FlavorResourceQuantities) (bool, resources.FlavorResourceQuantities) {
 	fits := true
 	remainingRequests := make(resources.FlavorResourceQuantities, len(requests))
 	for fr, v := range requests {
@@ -218,7 +223,7 @@ func QuantitiesFitInQuota(node hierarchicalResourceNode, requests resources.Flav
 
 // IsWithinNominalInResources returns whether or not, the node quota usage exceeds its
 // nominal quota in any resource flavor out of a set of resource flavours.
-func IsWithinNominalInResources(node hierarchicalResourceNode, frs sets.Set[resources.FlavorResource]) bool {
+func IsWithinNominalInResources(node flatResourceNode, frs sets.Set[resources.FlavorResource]) bool {
 	for fr := range frs {
 		if node.getResourceNode().Usage[fr] > node.getResourceNode().SubtreeQuota[fr] {
 			return false
diff --git a/pkg/controller/jobs/rayjob/rayjob_webhook.go b/pkg/controller/jobs/rayjob/rayjob_webhook.go
index 82fc55b48..78bf7b321 100644
--- a/pkg/controller/jobs/rayjob/rayjob_webhook.go
+++ b/pkg/controller/jobs/rayjob/rayjob_webhook.go
@@ -32,7 +32,6 @@ import (
 	"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"
 	"sigs.k8s.io/kueue/pkg/queue"
 )
 
@@ -101,11 +100,11 @@ func (w *RayJobWebhook) validateCreate(job *rayv1.RayJob) field.ErrorList {
 	var allErrors field.ErrorList
 	kueueJob := (*RayJob)(job)
 
-	if w.manageJobsWithoutQueueName || jobframework.QueueName(kueueJob) != "" || features.Enabled(features.LocalQueueDefaulting) {
+	if w.manageJobsWithoutQueueName || jobframework.QueueName(kueueJob) != "" {
 		spec := &job.Spec
 		specPath := field.NewPath("spec")
 
-		// Should always delete the cluster after the sob has ended, otherwise it will continue to the queue's resources.
+		// Should always delete the cluster after the job has ended, otherwise it will continue to the queue's resources.
 		if !spec.ShutdownAfterJobFinishes {
 			allErrors = append(allErrors, field.Invalid(specPath.Child("shutdownAfterJobFinishes"), spec.ShutdownAfterJobFinishes, "a kueue managed job should delete the cluster after finishing"))
 		}
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index 2c9ca6c1b..ad9821c24 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -230,6 +230,7 @@ var defaultVersionedFeatureGates = map[featuregate.Feature]featuregate.Versioned
 	},
 	LocalQueueDefaulting: {
 		{Version: version.MustParse("0.10"), Default: false, PreRelease: featuregate.Alpha},
+		{Version: version.MustParse("0.12"), Default: true, PreRelease: featuregate.Beta},
 	},
 	TASProfileMostFreeCapacity: {
 		{Version: version.MustParse("0.11"), Default: false, PreRelease: featuregate.Deprecated},
diff --git a/site/content/en/docs/installation/_index.md b/site/content/en/docs/installation/_index.md
index 07657d2d5..0b5cfdbf6 100644
--- a/site/content/en/docs/installation/_index.md
+++ b/site/content/en/docs/installation/_index.md
@@ -285,7 +285,8 @@ spec:
 | `WorkloadResourceRequestsSummary`     | `false` | Alpha      | 0.9   | 0.9   |
 | `WorkloadResourceRequestsSummary`     | `true`  | Beta       | 0.10  | 0.10  |
 | `ManagedJobsNamespaceSelector`        | `true`  | Beta       | 0.10  |       |
-| `LocalQueueDefaulting`                | `false` | Alpha      | 0.10  |       |
+| `LocalQueueDefaulting`                | `false` | Alpha      | 0.10  | 0.11  |
+| `LocalQueueDefaulting`                | `true`  | Beta       | 0.12  |       |
 | `LocalQueueMetrics`                   | `false` | Alpha      | 0.10  |       |
 
 ### Feature gates for graduated or deprecated features
diff --git a/site/content/en/docs/tasks/manage/enforce_job_management/setup_default_local_queue.md b/site/content/en/docs/tasks/manage/enforce_job_management/setup_default_local_queue.md
index c4eb49912..b1d8aa20e 100644
--- a/site/content/en/docs/tasks/manage/enforce_job_management/setup_default_local_queue.md
+++ b/site/content/en/docs/tasks/manage/enforce_job_management/setup_default_local_queue.md
@@ -6,19 +6,24 @@ description: >
   Setup default LocalQueue to fullfil a queue label on jobs that submited without queue label.
 ---
 
-{{< feature-state state="alpha" for_version="v0.10" >}}
+{{< feature-state state="beta" for_version="v0.12" >}}
+
+{{% alert title="Note" color="primary" %}}
+
+`LocalQueueDefaulting` is a Beta feature that is enabled by default.
+
+You can disable it by setting the `LocalQueueDefaulting` feature gate. Check the [Installation](/docs/installation/#change-the-feature-gates-configuration) guide for details on feature gate configuration.
+{{% /alert %}}
 
 This page describes how to setup default LocalQueue to ensure that all Workloads submitted to a specific namespace are managed by Kueue,
 even if the `kueue.x-k8s.io/queue-name` label is not specified explicitly.
 
 ## Setup default LocalQueue
 
-LocalQueueDefaulting is an Alpha feature that allows the use of a LocalQueue with name `default` as the default LocalQueue
+LocalQueueDefaulting is an Beta feature that allows the use of a LocalQueue with name `default` as the default LocalQueue
 for workloads in the same namespace that do not have the `kueue.x-k8s.io/queue-name` label.
-The feature is gated by the `LocalQueueDefaulting` feature gate, and is disabled by default. To use this feature:
+The feature is gated by the `LocalQueueDefaulting` feature gate, and is enabled by default. To use this feature:
 
-- Enable the LocalQueueDefaulting feature gate. Refer to the [feature gates configuration](/docs/installation/#change-the-feature-gates-configuration)
-guide for details.
 - create a LocalQueue with the name `default` in a namespace.
 
 That's all! Now, to test the feature, create a Job in the same namespace. Observe that the Job is updated with the `kueue.x-k8s.io/queue-name: default` label.



Test patch:
diff --git a/pkg/controller/jobs/rayjob/rayjob_webhook_test.go b/pkg/controller/jobs/rayjob/rayjob_webhook_test.go
index 39387188a..01ca3fd02 100644
--- a/pkg/controller/jobs/rayjob/rayjob_webhook_test.go
+++ b/pkg/controller/jobs/rayjob/rayjob_webhook_test.go
@@ -43,7 +43,7 @@ var (
 	workloadPriorityClassNamePath = labelsPath.Key(constants.WorkloadPriorityClassLabel)
 )
 
-func TestValidateDefault(t *testing.T) {
+func TestDefault(t *testing.T) {
 	testcases := map[string]struct {
 		oldJob               *rayv1.RayJob
 		newJob               *rayv1.RayJob
@@ -115,7 +115,7 @@ func TestValidateDefault(t *testing.T) {
 			if tc.defaultLqExist {
 				if err := queueManager.AddLocalQueue(ctx, utiltesting.MakeLocalQueue("default", "default").
 					ClusterQueue("cluster-queue").Obj()); err != nil {
-					t.Fatalf("failed to create default local queue: %s", err)
+					t.Fatalf("failed to create default local queue: %v", err)
 				}
 			}
 			wh := &RayJobWebhook{
@@ -125,7 +125,7 @@ func TestValidateDefault(t *testing.T) {
 			}
 			result := tc.oldJob.DeepCopy()
 			if err := wh.Default(t.Context(), result); err != nil {
-				t.Errorf("unexpected Default() error: %s", err)
+				t.Errorf("unexpected Default() error: %v", err)
 			}
 			if diff := cmp.Diff(tc.newJob, result); diff != "" {
 				t.Errorf("Default() mismatch (-want +got):\n%s", diff)
@@ -139,9 +139,10 @@ func TestValidateCreate(t *testing.T) {
 	bigWorkerGroup := []rayv1.WorkerGroupSpec{worker, worker, worker, worker, worker, worker, worker, worker}
 
 	testcases := map[string]struct {
-		job       *rayv1.RayJob
-		manageAll bool
-		wantErr   error
+		job                  *rayv1.RayJob
+		manageAll            bool
+		wantErr              error
+		localQueueDefaulting bool
 	}{
 		"invalid unmanaged": {
 			job: testingrayutil.MakeJob("job", "ns").
@@ -149,6 +150,13 @@ func TestValidateCreate(t *testing.T) {
 				Obj(),
 			wantErr: nil,
 		},
+		"invalid unmanaged - local queue default": {
+			job: testingrayutil.MakeJob("job", "ns").
+				ShutdownAfterJobFinishes(false).
+				Obj(),
+			localQueueDefaulting: true,
+			wantErr:              nil,
+		},
 		"invalid managed - by config": {
 			job: testingrayutil.MakeJob("job", "ns").
 				ShutdownAfterJobFinishes(false).
@@ -284,6 +292,7 @@ func TestValidateCreate(t *testing.T) {
 			wh := &RayJobWebhook{
 				manageJobsWithoutQueueName: tc.manageAll,
 			}
+			features.SetFeatureGateDuringTest(t, features.LocalQueueDefaulting, tc.localQueueDefaulting)
 			_, result := wh.ValidateCreate(t.Context(), tc.job)
 			if diff := cmp.Diff(tc.wantErr, result); diff != "" {
 				t.Errorf("ValidateCreate() mismatch (-want +got):\n%s", diff)
diff --git a/pkg/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index 06dab3a74..50790237e 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -2655,7 +2655,68 @@ func TestFairPreemptions(t *testing.T) {
 			admitted: []kueue.Workload{
 				unitWl.Clone().Name("b1").SimpleReserveQuota("b", "default", now).Workload,
 			},
-			incoming: unitWl.Clone().Name("a1").SimpleReserveQuota("a", "default", now).Obj(),
+			incoming: unitWl.Clone().Name("a1").Obj(),
+			targetCQ: "a",
+			wantPreempted: sets.New(
+				targetKeyReason("/b1", kueue.InCohortFairSharingReason),
+			),
+		},
+		"cq with zero weight can reclaim nominal quota": {
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("a").
+					Cohort("ROOT").
+					FairWeight(resource.MustParse("0.0")).
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "1").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					Obj(),
+				utiltesting.MakeClusterQueue("b").
+					Cohort("ROOT").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "0").Obj()).
+					FairWeight(resource.MustParse("1.0")).
+					Obj(),
+			},
+			admitted: []kueue.Workload{
+				*unitWl.Clone().Name("b1").SimpleReserveQuota("b", "default", now).Obj(),
+			},
+			incoming: unitWl.Clone().Name("a1").Obj(),
+			targetCQ: "a",
+			wantPreempted: sets.New(
+				targetKeyReason("/b1", kueue.InCohortFairSharingReason),
+			),
+		},
+		"cohort with zero weight can reclaim nominal quota": {
+			clusterQueues: []*kueue.ClusterQueue{
+				utiltesting.MakeClusterQueue("a").
+					Cohort("A").
+					FairWeight(resource.MustParse("0.0")).
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "0").Obj()).
+					Preemption(kueue.ClusterQueuePreemption{
+						ReclaimWithinCohort: kueue.PreemptionPolicyAny,
+					}).
+					Obj(),
+				utiltesting.MakeClusterQueue("b").
+					Cohort("ROOT").
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "0").Obj()).
+					FairWeight(resource.MustParse("1.0")).
+					Obj(),
+			},
+			cohorts: []*kueuealpha.Cohort{
+				utiltesting.MakeCohort("A").
+					Parent("ROOT").
+					FairWeight(resource.MustParse("0.0")).
+					ResourceGroup(*utiltesting.MakeFlavorQuotas("default").
+						Resource(corev1.ResourceCPU, "1").Obj()).Obj(),
+			},
+			admitted: []kueue.Workload{
+				*unitWl.Clone().Name("b1").SimpleReserveQuota("b", "default", now).Obj(),
+			},
+			incoming: unitWl.Clone().Name("a1").Obj(),
 			targetCQ: "a",
 			wantPreempted: sets.New(
 				targetKeyReason("/b1", kueue.InCohortFairSharingReason),
diff --git a/test/e2e/config/default/controller_manager_config.yaml b/test/e2e/config/default/controller_manager_config.yaml
index 8295a62a7..bf08f2d89 100644
--- a/test/e2e/config/default/controller_manager_config.yaml
+++ b/test/e2e/config/default/controller_manager_config.yaml
@@ -33,3 +33,7 @@ integrations:
   - "deployment"
   - "statefulset"
   - "leaderworkerset.x-k8s.io/leaderworkerset"
+featureGates:
+  MultiKueueBatchJobWithManagedBy: true
+  TopologyAwareScheduling: true
+  LocalQueueMetrics: true
diff --git a/test/e2e/config/default/manager_e2e_patch.yaml b/test/e2e/config/default/manager_e2e_patch.yaml
index 007f65bcc..8cdf47c02 100644
--- a/test/e2e/config/default/manager_e2e_patch.yaml
+++ b/test/e2e/config/default/manager_e2e_patch.yaml
@@ -1,6 +1,3 @@
 - op: replace
   path: /spec/template/spec/containers/0/imagePullPolicy
   value: IfNotPresent
-- op: add
-  path: /spec/template/spec/containers/0/args/-
-  value: --feature-gates=MultiKueueBatchJobWithManagedBy=true,TopologyAwareScheduling=true,LocalQueueMetrics=true
diff --git a/test/e2e/customconfigs/managejobswithoutqueuename_test.go b/test/e2e/customconfigs/managejobswithoutqueuename_test.go
index ca8ad7206..1c6e06ba2 100644
--- a/test/e2e/customconfigs/managejobswithoutqueuename_test.go
+++ b/test/e2e/customconfigs/managejobswithoutqueuename_test.go
@@ -84,6 +84,103 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 		util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, defaultRf, true, util.LongTimeout)
 		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sClient, ns)
 	})
+
+	ginkgo.When("manageJobsWithoutQueueName=true and LocalQueueDefaulting=false", func() {
+		ginkgo.It("should suspend a job", func() {
+			var testJob, createdJob *batchv1.Job
+			var jobLookupKey types.NamespacedName
+
+			ginkgo.By("setting local queue defulting as false", func() {
+				updateKueueConfiguration(func(cfg *config.Configuration) {
+					cfg.FeatureGates = map[string]bool{"LocalQueueDefaulting": false}
+					cfg.ManageJobsWithoutQueueName = true
+				})
+			})
+
+			ginkgo.By("creating a default LocalQueue", func() {
+				localQueue = testing.MakeLocalQueue("default", ns.Name).ClusterQueue("cluster-queue").Obj()
+				util.MustCreate(ctx, k8sClient, localQueue)
+			})
+
+			ginkgo.By("creating an unsuspended job without a queue name", func() {
+				testJob = testingjob.MakeJob("test-job", ns.Name).
+					Suspend(false).
+					Image(util.E2eTestAgnHostImage, util.BehaviorExitFast).
+					Obj()
+				util.MustCreate(ctx, k8sClient, testJob)
+			})
+
+			ginkgo.By("verifying that the job does get suspended", func() {
+				createdJob = &batchv1.Job{}
+				jobLookupKey = types.NamespacedName{Name: testJob.Name, Namespace: ns.Name}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, jobLookupKey, createdJob)).Should(gomega.Succeed())
+					g.Expect(ptr.Deref(createdJob.Spec.Suspend, false)).To(gomega.BeTrue())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying no existence of kueue labels", func() {
+				jobLookupKey = types.NamespacedName{Name: testJob.Name, Namespace: ns.Name}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, jobLookupKey, createdJob)).Should(gomega.Succeed())
+					g.Expect(createdJob.Labels).ShouldNot(gomega.HaveKeyWithValue(controllerconstants.QueueLabel, "default"))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+			ginkgo.By("setting feature gates back to its original state", func() {
+				updateKueueConfiguration(func(cfg *config.Configuration) {
+					cfg.ManageJobsWithoutQueueName = true
+				})
+			})
+		})
+	})
+
+	ginkgo.When("manageJobsWithoutQueueName=true and LocalQueueDefaulting=true", func() {
+		ginkgo.It("should not suspend a job", func() {
+			var testJob, createdJob *batchv1.Job
+			var jobLookupKey types.NamespacedName
+
+			ginkgo.By("creating a default LocalQueue", func() {
+				localQueue = testing.MakeLocalQueue("default", ns.Name).ClusterQueue("cluster-queue").Obj()
+				util.MustCreate(ctx, k8sClient, localQueue)
+			})
+
+			ginkgo.By("creating an unsuspended job without a queue name", func() {
+				testJob = testingjob.MakeJob("test-job", ns.Name).
+					Suspend(false).
+					Image(util.E2eTestAgnHostImage, util.BehaviorExitFast).
+					Obj()
+				util.MustCreate(ctx, k8sClient, testJob)
+			})
+
+			ginkgo.By("verifying that the job does not get suspended", func() {
+				createdJob = &batchv1.Job{}
+				jobLookupKey = types.NamespacedName{Name: testJob.Name, Namespace: ns.Name}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, jobLookupKey, createdJob)).Should(gomega.Succeed())
+					g.Expect(ptr.Deref(createdJob.Spec.Suspend, false)).To(gomega.BeFalse())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying existence of kueue labels", func() {
+				jobLookupKey = types.NamespacedName{Name: testJob.Name, Namespace: ns.Name}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, jobLookupKey, createdJob)).Should(gomega.Succeed())
+					g.Expect(createdJob.Labels).Should(gomega.HaveKeyWithValue(controllerconstants.QueueLabel, "default"))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying that the job has been admitted", func() {
+				wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(testJob.Name, testJob.UID), Namespace: ns.Name}
+				createdWorkload := &kueue.Workload{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
+					g.Expect(createdWorkload.Status.Admission).ShouldNot(gomega.BeNil())
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, createdWorkload)
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+
 	ginkgo.When("manageJobsWithoutQueueName=true", func() {
 		ginkgo.It("should suspend a job", func() {
 			var testJob, createdJob *batchv1.Job