Repository: kubernetes-sigs/kueue

Issue:
TAS: Implement e2e tests for RayJob with TAS
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:
E2e tests for RayJob with TAS

**Why is this needed**:

Discussed here as well: https://github.com/kubernetes-sigs/kueue/pull/3704#discussion_r1866116694

Hints from issue comments:
@PBundyra: /cc @mbobrovskyi @mszadkow: /assign@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@mimowo: /remove-lifecycle stale
It is being done as part of https://github.com/kubernetes-sigs/kueue/pull/4341

Solution patch:
diff --git a/Makefile-test.mk b/Makefile-test.mk
index a4b406120..85f55cdc7 100644
--- a/Makefile-test.mk
+++ b/Makefile-test.mk
@@ -154,6 +154,7 @@ run-test-tas-e2e-%: FORCE
 		JOBSET_VERSION=$(JOBSET_VERSION) KUBEFLOW_VERSION=$(KUBEFLOW_VERSION) KUBEFLOW_MPI_VERSION=$(KUBEFLOW_MPI_VERSION) \
 		APPWRAPPER_VERSION=$(APPWRAPPER_VERSION) \
 		LEADERWORKERSET_VERSION=$(LEADERWORKERSET_VERSION) \
+		KUBERAY_VERSION=$(KUBERAY_VERSION) \
 		KIND_CLUSTER_FILE="tas-kind-cluster.yaml" E2E_TARGET_FOLDER="tas" \
 		./hack/e2e-test.sh
 	$(PROJECT_DIR)/bin/ginkgo-top -i $(ARTIFACTS)/$@/e2e.json > $(ARTIFACTS)/$@/e2e-top.yaml
diff --git a/hack/e2e-test.sh b/hack/e2e-test.sh
index 4a041c07a..5376182c8 100755
--- a/hack/e2e-test.sh
+++ b/hack/e2e-test.sh
@@ -68,6 +68,9 @@ function kind_load {
     if [[ -n ${LEADERWORKERSET_VERSION:-} ]]; then
         install_lws "$KIND_CLUSTER_NAME"
     fi
+    if [[ -n ${KUBERAY_VERSION:-} ]]; then
+        install_kuberay "$KIND_CLUSTER_NAME"
+    fi
 }
 
 function kueue_deploy {
diff --git a/pkg/controller/jobs/rayjob/rayjob_controller.go b/pkg/controller/jobs/rayjob/rayjob_controller.go
index 6bb552342..9e13fd47b 100644
--- a/pkg/controller/jobs/rayjob/rayjob_controller.go
+++ b/pkg/controller/jobs/rayjob/rayjob_controller.go
@@ -144,11 +144,14 @@ func (j *RayJob) PodSets() ([]kueue.PodSet, error) {
 	// submitter Job
 	if j.Spec.SubmissionMode == rayv1.K8sJobMode {
 		submitterJobPodSet := kueue.PodSet{
-			Name:  submitterJobPodSetName,
-			Count: 1,
+			Name:     submitterJobPodSetName,
+			Count:    1,
+			Template: *getSubmitterTemplate(j),
 		}
 
-		submitterJobPodSet.Template = *getSubmitterTemplate(j)
+		// Create the TopologyRequest for the Submitter Job PodSet, based on the annotations
+		// in rayJob.Spec.SubmitterPodTemplate, which can be specified by the user.
+		submitterJobPodSet.TopologyRequest = jobframework.PodSetTopologyRequest(&submitterJobPodSet.Template.ObjectMeta, nil, nil, nil)
 		podSets = append(podSets, submitterJobPodSet)
 	}
 
diff --git a/pkg/resources/resource.go b/pkg/resources/resource.go
index 817e7db24..af0e691f3 100644
--- a/pkg/resources/resource.go
+++ b/pkg/resources/resource.go
@@ -17,6 +17,9 @@ limitations under the License.
 package resources
 
 import (
+	"encoding/json"
+	"fmt"
+
 	corev1 "k8s.io/api/core/v1"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
@@ -27,4 +30,16 @@ type FlavorResource struct {
 	Resource corev1.ResourceName
 }
 
+func (fr FlavorResource) String() string {
+	return fmt.Sprintf(`{"Flavor":"%s","Resource":"%s"}`, string(fr.Flavor), string(fr.Resource))
+}
+
 type FlavorResourceQuantities map[FlavorResource]int64
+
+func (q FlavorResourceQuantities) MarshalJSON() ([]byte, error) {
+	temp := make(map[string]int64, len(q))
+	for flavourResource, num := range q {
+		temp[flavourResource.String()] = num
+	}
+	return json.Marshal(temp)
+}
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 5cae76804..fd1566da2 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -253,7 +253,7 @@ func (p *Preemptor) applyPreemptionWithSSA(ctx context.Context, w *kueue.Workloa
 // fits.
 func minimalPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info, allowBorrowing bool, allowBorrowingBelowPriority *int32) []*Target {
 	if logV := preemptionCtx.log.V(5); logV.Enabled() {
-		logV.Info("Simulating preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption, "allowBorrowing", allowBorrowing, "allowBorrowingBelowPriority", allowBorrowingBelowPriority, "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
+		logV.Info("Simulating preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "allowBorrowing", allowBorrowing, "allowBorrowingBelowPriority", allowBorrowingBelowPriority, "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
 	}
 	// Simulate removing all candidates from the ClusterQueue and cohort.
 	var targets []*Target
@@ -359,7 +359,7 @@ func parseStrategies(s []config.PreemptionStrategy) []fsStrategy {
 
 func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, candidates []*workload.Info, allowBorrowingBelowPriority *int32) []*Target {
 	if logV := preemptionCtx.log.V(5); logV.Enabled() {
-		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption, "allowBorrowingBelowPriority", allowBorrowingBelowPriority, "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
+		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "allowBorrowingBelowPriority", allowBorrowingBelowPriority, "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
 	}
 	requests := preemptionCtx.requests
 	cqHeap := cqHeapFromCandidates(candidates, false, preemptionCtx.snapshot)



Test patch:
diff --git a/pkg/controller/jobs/rayjob/rayjob_controller_test.go b/pkg/controller/jobs/rayjob/rayjob_controller_test.go
index 4d08866a1..d70d5aa59 100644
--- a/pkg/controller/jobs/rayjob/rayjob_controller_test.go
+++ b/pkg/controller/jobs/rayjob/rayjob_controller_test.go
@@ -184,6 +184,119 @@ func TestPodSets(t *testing.T) {
 				}
 			},
 		},
+		"with submitter topology annotation not specified": {
+			rayJob: (*RayJob)(testingrayutil.MakeJob("rayjob", "ns").
+				WithSubmissionMode(rayv1.K8sJobMode).
+				WithHeadGroupSpec(
+					rayv1.HeadGroupSpec{
+						Template: corev1.PodTemplateSpec{
+							ObjectMeta: metav1.ObjectMeta{
+								Annotations: map[string]string{
+									kueuealpha.PodSetRequiredTopologyAnnotation: "cloud.com/rack",
+								},
+							},
+							Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "head_c"}}},
+						},
+					},
+				).
+				WithWorkerGroups(
+					rayv1.WorkerGroupSpec{
+						GroupName: "group1",
+						Replicas:  ptr.To[int32](3),
+						Template: corev1.PodTemplateSpec{
+							ObjectMeta: metav1.ObjectMeta{
+								Annotations: map[string]string{
+									kueuealpha.PodSetPreferredTopologyAnnotation: "cloud.com/block",
+								},
+							},
+							Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "group1_c"}}},
+						},
+					},
+				).
+				Obj()),
+			wantPodSets: func(rayJob *RayJob) []kueue.PodSet {
+				return []kueue.PodSet{
+					*utiltesting.MakePodSet(headGroupPodSetName, 1).
+						PodSpec(*rayJob.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.DeepCopy()).
+						Annotations(rayJob.Spec.RayClusterSpec.HeadGroupSpec.Template.Annotations).
+						RequiredTopologyRequest("cloud.com/rack").
+						Obj(),
+					*utiltesting.MakePodSet(kueue.NewPodSetReference(rayJob.Spec.RayClusterSpec.WorkerGroupSpecs[0].GroupName), 3).
+						PodSpec(*rayJob.Spec.RayClusterSpec.WorkerGroupSpecs[0].Template.Spec.DeepCopy()).
+						Annotations(rayJob.Spec.RayClusterSpec.WorkerGroupSpecs[0].Template.Annotations).
+						PreferredTopologyRequest("cloud.com/block").
+						Obj(),
+					*utiltesting.MakePodSet("submitter", 1).
+						PodSpec(*getSubmitterTemplate(rayJob).Spec.DeepCopy()).
+						Obj(),
+				}
+			},
+		},
+		"with submitter topology annotation specified": {
+			rayJob: (*RayJob)(testingrayutil.MakeJob("rayjob", "ns").
+				WithSubmissionMode(rayv1.K8sJobMode).
+				WithHeadGroupSpec(
+					rayv1.HeadGroupSpec{
+						Template: corev1.PodTemplateSpec{
+							ObjectMeta: metav1.ObjectMeta{
+								Annotations: map[string]string{
+									kueuealpha.PodSetRequiredTopologyAnnotation: "cloud.com/rack",
+								},
+							},
+							Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "head_c"}}},
+						},
+					},
+				).
+				WithWorkerGroups(
+					rayv1.WorkerGroupSpec{
+						GroupName: "group1",
+						Replicas:  ptr.To[int32](3),
+						Template: corev1.PodTemplateSpec{
+							ObjectMeta: metav1.ObjectMeta{
+								Annotations: map[string]string{
+									kueuealpha.PodSetPreferredTopologyAnnotation: "cloud.com/block",
+								},
+							},
+							Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "group1_c"}}},
+						},
+					},
+				).
+				WithSubmitterPodTemplate(corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name: "submitter_c",
+							},
+						},
+						RestartPolicy: corev1.RestartPolicyNever,
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{
+							kueuealpha.PodSetRequiredTopologyAnnotation: "cloud.com/block",
+						},
+					},
+				}).
+				Obj()),
+			wantPodSets: func(rayJob *RayJob) []kueue.PodSet {
+				return []kueue.PodSet{
+					*utiltesting.MakePodSet(headGroupPodSetName, 1).
+						PodSpec(*rayJob.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.DeepCopy()).
+						Annotations(rayJob.Spec.RayClusterSpec.HeadGroupSpec.Template.Annotations).
+						RequiredTopologyRequest("cloud.com/rack").
+						Obj(),
+					*utiltesting.MakePodSet(kueue.NewPodSetReference(rayJob.Spec.RayClusterSpec.WorkerGroupSpecs[0].GroupName), 3).
+						PodSpec(*rayJob.Spec.RayClusterSpec.WorkerGroupSpecs[0].Template.Spec.DeepCopy()).
+						Annotations(rayJob.Spec.RayClusterSpec.WorkerGroupSpecs[0].Template.Annotations).
+						PreferredTopologyRequest("cloud.com/block").
+						Obj(),
+					*utiltesting.MakePodSet("submitter", 1).
+						PodSpec(*rayJob.Spec.SubmitterPodTemplate.Spec.DeepCopy()).
+						Annotations(rayJob.Spec.SubmitterPodTemplate.Annotations).
+						RequiredTopologyRequest("cloud.com/block").
+						Obj(),
+				}
+			},
+		},
 		"with NumOfHosts > 1": {
 			rayJob: (*RayJob)(testingrayutil.MakeJob("rayjob", "ns").
 				WithHeadGroupSpec(
diff --git a/pkg/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index 71a741356..1dd023257 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -295,16 +295,25 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("mid", "").
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -326,16 +335,25 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("mid", "").
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -358,11 +376,17 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("mid", "").
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -383,11 +407,17 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("mid", "").
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -407,16 +437,25 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "1000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "1000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("mid", "").
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "1000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "1000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -438,16 +477,25 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "1000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "1000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("mid", "").
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -469,16 +517,25 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low", "").
 					Priority(-1).
 					Request(corev1.ResourceMemory, "2Gi").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "alpha", "2Gi").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "alpha", "2Gi").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("mid", "").
 					Request(corev1.ResourceMemory, "1Gi").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "beta", "1Gi").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "beta", "1Gi").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("high", "").
 					Priority(1).
 					Request(corev1.ResourceMemory, "1Gi").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "beta", "1Gi").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "beta", "1Gi").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -505,16 +562,25 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-mid", "").
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "6").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "6000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "6000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -573,12 +639,18 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c1-high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-1", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -599,17 +671,26 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c1-high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-1", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-2", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -630,17 +711,26 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-1", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-high-2", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -667,17 +757,26 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c1-med", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-1", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -699,17 +798,26 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("d1-med", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("d1").Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("d1").Assignment(corev1.ResourceCPU, "default", "4").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("d1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("d1").Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("d1").Assignment(corev1.ResourceCPU, "default", "4").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("d2-low-1", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("d2").Assignment(corev1.ResourceCPU, "default", "4").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("d2").Assignment(corev1.ResourceCPU, "default", "4").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -731,22 +839,34 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c1-med", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-1", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "5").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "5000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "5000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-2", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "1000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "1000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low-3", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "1000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "1000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -773,12 +893,18 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("l1-med", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("l1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("l1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("l1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("l1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("l1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -799,15 +925,24 @@ func TestPreemption(t *testing.T) {
 			admitted: []kueue.Workload{
 				*utiltesting.MakeWorkload("c1", "").
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-1", "").
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-2", "").
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -826,16 +961,25 @@ func TestPreemption(t *testing.T) {
 			admitted: []kueue.Workload{
 				*utiltesting.MakeWorkload("c1-1", "").
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c1-2", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2", "").
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -856,20 +1000,32 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c1-mid", "").
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("c2-mid", "").
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -890,7 +1046,10 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("c2-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("c2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -911,12 +1070,18 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("low-alpha", "").
 					Priority(-1).
 					Request(corev1.ResourceMemory, "2Gi").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "alpha", "2Gi").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "alpha", "2Gi").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("low-beta", "").
 					Priority(-1).
 					Request(corev1.ResourceMemory, "2Gi").
-					ReserveQuota(utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "beta", "2Gi").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("standalone").Assignment(corev1.ResourceMemory, "beta", "2Gi").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -960,24 +1125,28 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("wl1", "").
 					Priority(2).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("preventStarvation").Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("preventStarvation").Assignment(corev1.ResourceCPU, "default", "2").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("wl2", "").
 					Priority(1).
 					Creation(now).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("preventStarvation").Assignment(corev1.ResourceCPU, "default", "2").Obj()).
-					SetOrReplaceCondition(metav1.Condition{
-						Type:               kueue.WorkloadQuotaReserved,
-						Status:             metav1.ConditionTrue,
-						LastTransitionTime: metav1.NewTime(now.Add(time.Second)),
-					}).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("preventStarvation").Assignment(corev1.ResourceCPU, "default", "2").Obj(),
+						now.Add(time.Second),
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("wl3", "").
 					Priority(1).
 					Creation(now).
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("preventStarvation").Assignment(corev1.ResourceCPU, "default", "2").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("preventStarvation").Assignment(corev1.ResourceCPU, "default", "2").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1010,12 +1179,18 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("a_best_effort_low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "10").
-					ReserveQuota(utiltesting.MakeAdmission("a_best_effort").Assignment(corev1.ResourceCPU, "default", "10").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a_best_effort").Assignment(corev1.ResourceCPU, "default", "10").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b_best_effort_low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b_best_effort").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b_best_effort").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1036,7 +1211,10 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("b_standard", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "10").
-					ReserveQuota(utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "10000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "10000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1058,7 +1236,10 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("b_standard", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "13").
-					ReserveQuota(utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "13000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "13000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1081,7 +1262,10 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("a_standard", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "13").
-					ReserveQuota(utiltesting.MakeAdmission("a_standard").Assignment(corev1.ResourceCPU, "default", "13000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a_standard").Assignment(corev1.ResourceCPU, "default", "13000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1102,22 +1286,34 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("a_standard_1", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "10").
-					ReserveQuota(utiltesting.MakeAdmission("a_standard").Assignment(corev1.ResourceCPU, "default", "10").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a_standard").Assignment(corev1.ResourceCPU, "default", "10").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a_standard_2", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b_standard_1", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b_standard_2", "").
 					Priority(2).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1139,22 +1335,34 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("b_standard_high", "").
 					Priority(2).
 					Request(corev1.ResourceCPU, "10").
-					ReserveQuota(utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "10").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "10").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b_standard_mid", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b_standard").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a_best_effort_low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a_best_effort").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a_best_effort").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a_best_effort_lower", "").
 					Priority(-2).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a_best_effort").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a_best_effort").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1176,16 +1384,25 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("lend1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("lend1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("lend2-mid", "").
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("lend2-high", "").
 					Priority(1).
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1207,20 +1424,32 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("lend1-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("lend1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend1").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("lend1-mid", "").
 					Request(corev1.ResourceCPU, "2").
-					ReserveQuota(utiltesting.MakeAdmission("lend1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend1").Assignment(corev1.ResourceCPU, "default", "2000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("lend2-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "3").
-					ReserveQuota(utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "3000m").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("lend2-mid", "").
 					Request(corev1.ResourceCPU, "4").
-					ReserveQuota(utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "4000m").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1241,7 +1470,10 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("lend2-low", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "10").
-					ReserveQuota(utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "10").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("lend2").Assignment(corev1.ResourceCPU, "default", "10").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1262,32 +1494,50 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("a1", "").
 					Priority(-2).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a2", "").
 					Priority(-2).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a3", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b1", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b2", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b3", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1310,37 +1560,55 @@ func TestPreemption(t *testing.T) {
 					Priority(-2).
 					Request(corev1.ResourceCPU, "1").
 					Request(corev1.ResourceMemory, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a2", "").
 					Priority(-2).
 					Request(corev1.ResourceCPU, "1").
 					Request(corev1.ResourceMemory, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a3", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
 					Request(corev1.ResourceMemory, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b1", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
 					Request(corev1.ResourceMemory, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b2", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
 					Request(corev1.ResourceMemory, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b3", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
 					Request(corev1.ResourceMemory, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Assignment(corev1.ResourceMemory, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1367,32 +1635,50 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("a1", "").
 					Priority(-2).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a2", "").
 					Priority(-2).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("a3", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b1", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b2", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b3", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1419,32 +1705,50 @@ func TestPreemption(t *testing.T) {
 				*utiltesting.MakeWorkload("a1", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("a").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b1", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b2", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b3", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b4", "").
 					Priority(0).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 				*utiltesting.MakeWorkload("b5", "").
 					Priority(-1).
 					Request(corev1.ResourceCPU, "1").
-					ReserveQuota(utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("b").Assignment(corev1.ResourceCPU, "default", "1").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("in", "").
@@ -1485,7 +1789,10 @@ func TestPreemption(t *testing.T) {
 			admitted: []kueue.Workload{
 				*utiltesting.MakeWorkload("to-be-preempted", "").
 					Request(corev1.ResourceCPU, "5").
-					ReserveQuota(utiltesting.MakeAdmission("cq-right").Assignment(corev1.ResourceCPU, "default", "5").Obj()).
+					ReserveQuotaAt(
+						utiltesting.MakeAdmission("cq-right").Assignment(corev1.ResourceCPU, "default", "5").Obj(),
+						now,
+					).
 					Obj(),
 			},
 			incoming: utiltesting.MakeWorkload("incoming", "").
@@ -2092,21 +2399,22 @@ func TestCandidatesOrdering(t *testing.T) {
 	now := time.Now()
 	candidates := []*workload.Info{
 		workload.NewInfo(utiltesting.MakeWorkload("high", "").
-			ReserveQuota(utiltesting.MakeAdmission("self").Obj()).
+			ReserveQuotaAt(utiltesting.MakeAdmission("self").Obj(), now).
 			Priority(10).
 			Obj()),
 		workload.NewInfo(utiltesting.MakeWorkload("low", "").
-			ReserveQuota(utiltesting.MakeAdmission("self").Obj()).
+			ReserveQuotaAt(utiltesting.MakeAdmission("self").Obj(), now).
 			Priority(-10).
 			Obj()),
 		workload.NewInfo(utiltesting.MakeWorkload("other", "").
-			ReserveQuota(utiltesting.MakeAdmission("other").Obj()).
+			ReserveQuotaAt(utiltesting.MakeAdmission("other").Obj(), now).
 			Priority(10).
 			Obj()),
 		workload.NewInfo(utiltesting.MakeWorkload("evicted", "").
 			SetOrReplaceCondition(metav1.Condition{
-				Type:   kueue.WorkloadEvicted,
-				Status: metav1.ConditionTrue,
+				Type:               kueue.WorkloadEvicted,
+				Status:             metav1.ConditionTrue,
+				LastTransitionTime: metav1.NewTime(now),
 			}).
 			Obj()),
 		workload.NewInfo(utiltesting.MakeWorkload("old-a", "").
@@ -2118,12 +2426,7 @@ func TestCandidatesOrdering(t *testing.T) {
 			ReserveQuotaAt(utiltesting.MakeAdmission("self").Obj(), now).
 			Obj()),
 		workload.NewInfo(utiltesting.MakeWorkload("current", "").
-			ReserveQuota(utiltesting.MakeAdmission("self").Obj()).
-			SetOrReplaceCondition(metav1.Condition{
-				Type:               kueue.WorkloadQuotaReserved,
-				Status:             metav1.ConditionTrue,
-				LastTransitionTime: metav1.NewTime(now.Add(time.Second)),
-			}).
+			ReserveQuotaAt(utiltesting.MakeAdmission("self").Obj(), now.Add(time.Second)).
 			Obj()),
 	}
 	sort.Slice(candidates, candidatesOrdering(candidates, "self", now))
diff --git a/test/e2e/customconfigs/skipjobswithoutqueuename_test.go b/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
index a51bbade6..3abaccf6d 100644
--- a/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
+++ b/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
@@ -251,9 +251,9 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, podLookupKey, createdPod)).Should(gomega.Succeed())
 					g.Expect(createdPod.Spec.SchedulingGates).Should(gomega.BeComparableTo([]corev1.PodSchedulingGate{
-						{Name: "kueue.x-k8s.io/admission"},
+						{Name: podcontroller.SchedulingGateName},
 					}))
-					g.Expect(createdPod.ObjectMeta.Labels).To(gomega.BeComparableTo(map[string]string{constants.ManagedByKueueLabelKey: constants.ManagedByKueueLabelValue}))
+					g.Expect(createdPod.Labels).Should(gomega.HaveKeyWithValue(constants.ManagedByKueueLabelKey, constants.ManagedByKueueLabelValue))
 					g.Expect(createdPod.Finalizers).Should(gomega.ContainElement(podcontroller.PodFinalizer))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
@@ -305,11 +305,12 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 					pods := &corev1.PodList{}
 					g.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Namespace),
 						client.MatchingLabels(testDeploy.Spec.Selector.MatchLabels))).To(gomega.Succeed())
+					g.Expect(pods.Items).Should(gomega.HaveLen(2))
 					for _, pod := range pods.Items {
 						g.Expect(pod.Spec.SchedulingGates).Should(gomega.BeComparableTo([]corev1.PodSchedulingGate{
-							{Name: "kueue.x-k8s.io/admission"},
+							{Name: podcontroller.SchedulingGateName},
 						}))
-						g.Expect(pod.ObjectMeta.Labels).To(gomega.BeComparableTo(map[string]string{constants.ManagedByKueueLabelKey: constants.ManagedByKueueLabelValue}))
+						g.Expect(pod.Labels).Should(gomega.HaveKeyWithValue(constants.ManagedByKueueLabelKey, constants.ManagedByKueueLabelValue))
 						g.Expect(pod.Finalizers).Should(gomega.ContainElement(podcontroller.PodFinalizer))
 					}
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
@@ -368,11 +369,13 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 					pods := &corev1.PodList{}
 					g.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Namespace),
 						client.MatchingLabels(testSts.Spec.Selector.MatchLabels))).To(gomega.Succeed())
+					// If the first pod can't be scheduled, the second won't be created
+					g.Expect(pods.Items).Should(gomega.HaveLen(1))
 					for _, pod := range pods.Items {
 						g.Expect(pod.Spec.SchedulingGates).Should(gomega.BeComparableTo([]corev1.PodSchedulingGate{
 							{Name: podcontroller.SchedulingGateName},
 						}))
-						g.Expect(pod.ObjectMeta.Labels).To(gomega.BeComparableTo(map[string]string{constants.ManagedByKueueLabelKey: constants.ManagedByKueueLabelValue}))
+						g.Expect(pod.Labels).Should(gomega.HaveKeyWithValue(constants.ManagedByKueueLabelKey, constants.ManagedByKueueLabelValue))
 						g.Expect(pod.Finalizers).Should(gomega.ContainElement(podcontroller.PodFinalizer))
 					}
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index 6f91d4b0e..098a02729 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -18,10 +18,8 @@ package mke2e
 
 import (
 	"fmt"
-	"os"
 	"os/exec"
 	"regexp"
-	"runtime"
 
 	"github.com/google/go-cmp/cmp/cmpopts"
 	kfmpi "github.com/kubeflow/mpi-operator/pkg/apis/kubeflow/v2beta1"
@@ -765,7 +763,7 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 		})
 
 		ginkgo.It("Should run a RayJob on worker if admitted", func() {
-			kuberayTestImage := getKuberayTestImage()
+			kuberayTestImage := util.GetKuberayTestImage()
 			// Since it requires 1.5 CPU, this job can only be admitted in worker 1.
 			rayjob := testingrayjob.MakeJob("rayjob1", managerNs.Name).
 				Suspend(true).
@@ -793,7 +791,7 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 					g.Expect(createdRayJob.Status.JobDeploymentStatus).To(gomega.Equal(rayv1.JobDeploymentStatusComplete))
 					finishReasonMessage := "Job finished successfully."
 					checkFinishStatusCondition(g, wlLookupKey, finishReasonMessage)
-				}, 5*util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				}, util.VeryLongTimeout, util.Interval).Should(gomega.Succeed())
 			})
 
 			ginkgo.By("Checking no objects are left in the worker clusters and the RayJob is completed", func() {
@@ -809,7 +807,7 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 		})
 
 		ginkgo.It("Should run a RayCluster on worker if admitted", func() {
-			kuberayTestImage := getKuberayTestImage()
+			kuberayTestImage := util.GetKuberayTestImage()
 			// Since it requires 1.5 CPU, this job can only be admitted in worker 1.
 			raycluster := testingraycluster.MakeCluster("raycluster1", managerNs.Name).
 				Suspend(true).
@@ -967,20 +965,6 @@ func checkFinishStatusCondition(g gomega.Gomega, wlLookupKey types.NamespacedNam
 	}, util.IgnoreConditionTimestampsAndObservedGeneration))
 }
 
-func getKuberayTestImage() string {
-	var (
-		kuberayTestImage string
-		found            bool
-	)
-	if runtime.GOARCH == "arm64" {
-		kuberayTestImage, found = os.LookupEnv("KUBERAY_RAY_IMAGE_ARM")
-	} else {
-		kuberayTestImage, found = os.LookupEnv("KUBERAY_RAY_IMAGE")
-	}
-	gomega.Expect(found).To(gomega.BeTrue())
-	return kuberayTestImage
-}
-
 func ensurePodWorkloadsRunning(deployment *appsv1.Deployment, managerNs corev1.Namespace, multiKueueAc *kueue.AdmissionCheck, kubernetesClients map[string]client.Client) {
 	// Given the unpredictable nature of where the deployment pods run this function gathers the workload of a Pod first
 	// it then gets the Pod's assigned cluster from the admission check message and uses the appropriate client to ensure the Pod is running
diff --git a/test/e2e/tas/rayjob_test.go b/test/e2e/tas/rayjob_test.go
new file mode 100644
index 000000000..22fa4c497
--- /dev/null
+++ b/test/e2e/tas/rayjob_test.go
@@ -0,0 +1,236 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tase2e
+
+import (
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/utils/ptr"
+	"k8s.io/utils/set"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/util/testing"
+	testingrayjob "sigs.k8s.io/kueue/pkg/util/testingjobs/rayjob"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var _ = ginkgo.Describe("TopologyAwareScheduling for RayJob", ginkgo.Ordered, func() {
+	var (
+		ns           *corev1.Namespace
+		topology     *kueuealpha.Topology
+		tasFlavor    *kueue.ResourceFlavor
+		clusterQueue *kueue.ClusterQueue
+		localQueue   *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeEach(func() {
+		ns = &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				GenerateName: "e2e-tas-rayjob-",
+			},
+		}
+		gomega.Expect(k8sClient.Create(ctx, ns)).To(gomega.Succeed())
+
+		topology = testing.MakeDefaultThreeLevelTopology("datacenter")
+		gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
+
+		tasFlavor = testing.MakeResourceFlavor("tas-flavor").
+			NodeLabel(tasNodeGroupLabel, instanceType).
+			TopologyName(topology.Name).
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, tasFlavor)).Should(gomega.Succeed())
+
+		clusterQueue = testing.MakeClusterQueue("cluster-queue").
+			ResourceGroup(
+				*testing.MakeFlavorQuotas(tasFlavor.Name).
+					Resource(corev1.ResourceCPU, "1").
+					Resource(corev1.ResourceMemory, "8Gi").
+					Resource(extraResource, "8").
+					Obj(),
+			).
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, clusterQueue)).Should(gomega.Succeed())
+		util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQueue)
+
+		localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+		gomega.Expect(k8sClient.Create(ctx, localQueue)).Should(gomega.Succeed())
+		util.ExpectLocalQueuesToBeActive(ctx, k8sClient, localQueue)
+	})
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteAllRayJobsInNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
+	})
+
+	ginkgo.When("Creating a RayJob", func() {
+		ginkgo.It("Should place pods based on the ranks-ordering", func() {
+			const (
+				headReplicas     = 1
+				workerReplicas   = 3
+				submitterReplica = 1
+			)
+			numPods := headReplicas + workerReplicas + submitterReplica
+			kuberayTestImage := util.GetKuberayTestImage()
+			rayjob := testingrayjob.MakeJob("ranks-ray", ns.Name).
+				Queue(localQueue.Name).
+				WithSubmissionMode(rayv1.K8sJobMode).
+				Entrypoint("python -c \"import ray; ray.init(); print(ray.cluster_resources())\"").
+				WithHeadGroupSpec(
+					rayv1.HeadGroupSpec{
+						RayStartParams: map[string]string{},
+						Template: corev1.PodTemplateSpec{
+							Spec: corev1.PodSpec{
+								RestartPolicy: corev1.RestartPolicyOnFailure,
+								Containers: []corev1.Container{
+									{
+										Name: "head-container",
+										Resources: corev1.ResourceRequirements{
+											Requests: corev1.ResourceList{
+												corev1.ResourceCPU: resource.MustParse("100m"),
+											},
+											Limits: corev1.ResourceList{
+												corev1.ResourceCPU: resource.MustParse("100m"),
+											},
+										},
+									},
+								},
+							},
+							ObjectMeta: metav1.ObjectMeta{
+								Annotations: map[string]string{
+									kueuealpha.PodSetPreferredTopologyAnnotation: testing.DefaultRackTopologyLevel,
+								},
+							},
+						},
+					},
+				).
+				WithWorkerGroups(
+					rayv1.WorkerGroupSpec{
+						GroupName:      "workers-group-0",
+						Replicas:       ptr.To[int32](workerReplicas),
+						MinReplicas:    ptr.To[int32](workerReplicas),
+						MaxReplicas:    ptr.To[int32](10),
+						RayStartParams: map[string]string{},
+						Template: corev1.PodTemplateSpec{
+							Spec: corev1.PodSpec{
+								RestartPolicy: corev1.RestartPolicyOnFailure,
+								Containers: []corev1.Container{
+									{
+										Name: "head-container",
+										Resources: corev1.ResourceRequirements{
+											Requests: corev1.ResourceList{
+												extraResource: resource.MustParse("1"),
+											},
+											Limits: corev1.ResourceList{
+												extraResource: resource.MustParse("1"),
+											},
+										},
+									},
+								},
+							},
+							ObjectMeta: metav1.ObjectMeta{
+								Annotations: map[string]string{
+									kueuealpha.PodSetPreferredTopologyAnnotation: testing.DefaultBlockTopologyLevel,
+								},
+							},
+						},
+					},
+				).
+				WithSubmitterPodTemplate(corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:  "ranks-ray-submitter",
+								Image: kuberayTestImage,
+								Resources: corev1.ResourceRequirements{
+									Requests: corev1.ResourceList{
+										corev1.ResourceCPU: resource.MustParse("50m"),
+									},
+									Limits: corev1.ResourceList{
+										corev1.ResourceCPU: resource.MustParse("50m"),
+									},
+								},
+							},
+						},
+						RestartPolicy: corev1.RestartPolicyNever,
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{
+							kueuealpha.PodSetRequiredTopologyAnnotation: testing.DefaultBlockTopologyLevel,
+						},
+					},
+				}).
+				Image(rayv1.HeadNode, kuberayTestImage, []string{}).
+				Image(rayv1.WorkerNode, kuberayTestImage, []string{}).
+				Obj()
+
+			gomega.Expect(k8sClient.Create(ctx, rayjob)).Should(gomega.Succeed())
+
+			ginkgo.By("RayJob is unsuspended", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(rayjob), rayjob)).To(gomega.Succeed())
+					g.Expect(rayjob.Spec.Suspend).Should(gomega.BeFalse())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			pods := &corev1.PodList{}
+			ginkgo.By("ensure all pods are created", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.InNamespace(rayjob.Namespace))).To(gomega.Succeed())
+					g.Expect(pods.Items).Should(gomega.HaveLen(numPods))
+					// The timeout is long to ensure all cluster pods are up and running.
+					// This is because the Ray image takes long time (around 170s on the CI)
+					// to load and then to sync with head.
+				}, util.VeryLongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("ensure all pods are scheduled", func() {
+				listOpts := &client.ListOptions{
+					FieldSelector: fields.OneTermNotEqualSelector("spec.nodeName", ""),
+				}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Name), listOpts)).To(gomega.Succeed())
+					g.Expect(pods.Items).Should(gomega.HaveLen(numPods))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verify the assignment of pods are as expected with rank-based ordering", func() {
+				gomega.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Name))).To(gomega.Succeed())
+				gotAssignment := readAssignedNodes(pods.Items)
+				gomega.Expect(gotAssignment).Should(gomega.HaveLen(workerReplicas))
+			})
+		})
+	})
+})
+
+func readAssignedNodes(pods []corev1.Pod) set.Set[string] {
+	assignment := set.Set[string]{}
+	for _, pod := range pods {
+		if role := pod.Labels["ray.io/node-type"]; role == "worker" {
+			assignment = assignment.Insert(pod.Spec.NodeName)
+		}
+	}
+	return assignment
+}
diff --git a/test/e2e/tas/suite_test.go b/test/e2e/tas/suite_test.go
index 9a3491386..5a3498a42 100644
--- a/test/e2e/tas/suite_test.go
+++ b/test/e2e/tas/suite_test.go
@@ -62,6 +62,7 @@ var _ = ginkgo.BeforeSuite(func() {
 	util.WaitForKubeFlowTrainingOperatorAvailability(ctx, k8sClient)
 	util.WaitForKubeFlowMPIOperatorAvailability(ctx, k8sClient)
 	util.WaitForAppWrapperAvailability(ctx, k8sClient)
+	util.WaitForKubeRayOperatorAvailability(ctx, k8sClient)
 	ginkgo.GinkgoLogr.Info(
 		"Kueue, JobSet, KubeFlow Training and KubeFlow MPI operators are available in the cluster",
 		"waitingTime", time.Since(waitForAvailableStart),
diff --git a/test/util/constants.go b/test/util/constants.go
index 7f8f71a8d..216abfe73 100644
--- a/test/util/constants.go
+++ b/test/util/constants.go
@@ -32,6 +32,9 @@ const (
 	// LongTimeout is meant for E2E tests when waiting for complex operations
 	// such as running pods to completion.
 	LongTimeout = 45 * time.Second
+	// VeryLongTimeout is meant for E2E tests involving Ray which starts ray-project images (over 2GB)
+	// and also synchronizes the cluster before it can be used
+	VeryLongTimeout = 3 * time.Minute
 	// StartUpTimeout is meant to be used for waiting for Kueue to startup, given
 	// that cert updates can take up to 3 minutes to propagate to the filesystem.
 	// Taken into account that after the certificates are ready, all Kueue's components
diff --git a/test/util/e2e.go b/test/util/e2e.go
index 5ba65e5fa..4b4877c6d 100644
--- a/test/util/e2e.go
+++ b/test/util/e2e.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"runtime"
 	"time"
 
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -273,3 +274,17 @@ func WaitForActivePodsAndTerminate(ctx context.Context, k8sClient client.Client,
 		}
 	}
 }
+
+func GetKuberayTestImage() string {
+	var (
+		kuberayTestImage string
+		found            bool
+	)
+	if runtime.GOARCH == "arm64" {
+		kuberayTestImage, found = os.LookupEnv("KUBERAY_RAY_IMAGE_ARM")
+	} else {
+		kuberayTestImage, found = os.LookupEnv("KUBERAY_RAY_IMAGE")
+	}
+	gomega.Expect(found).To(gomega.BeTrue())
+	return kuberayTestImage
+}
diff --git a/test/util/util.go b/test/util/util.go
index 33f6a187e..c0451d3e5 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -35,6 +35,7 @@ import (
 	"github.com/onsi/gomega"
 	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	"github.com/prometheus/client_golang/prometheus"
+	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	zaplog "go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	appsv1 "k8s.io/api/apps/v1"
@@ -161,6 +162,10 @@ func DeleteAllAppWrappersInNamespace(ctx context.Context, c client.Client, ns *c
 	return deleteAllObjectsInNamespace(ctx, c, ns, &awv1beta2.AppWrapper{})
 }
 
+func DeleteAllRayJobsInNamespace(ctx context.Context, c client.Client, ns *corev1.Namespace) error {
+	return deleteAllObjectsInNamespace(ctx, c, ns, &rayv1.RayJob{})
+}
+
 func DeleteAllPodsInNamespace(ctx context.Context, c client.Client, ns *corev1.Namespace) error {
 	return deleteAllPodsInNamespace(ctx, c, ns, 2)
 }