Repository: kubernetes-sigs/kueue
Pull request #209 (2022-04-12 13:24:09)

feat: sync workload condition when admitted
Signed-off-by: kerthcet <kerthcet@gmail.com>

<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide/first-contribution.md#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

#### What type of PR is this?
/kind feature

<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:

#### Which issue(s) this PR fixes: 
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes https://github.com/kubernetes-sigs/kueue/issues/201

#### Special notes for your reviewer:



k8s-ci-robot: Hi @kerthcet. Thanks for your PR.

I'm waiting for a [kubernetes-sigs](https://github.com/orgs/kubernetes-sigs/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkueue).

<details>

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

k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kueue/pull/209#" title="Author self-approved">kerthcet</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **ahg-g** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @ahg-g` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkueue).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-sigs/kueue/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["ahg-g"]} -->
kerthcet: cc @ahg-g 
alculquicondor: We'll probably want to leave this one after we cut the release.
I'm only waiting on #178 to merge to do it.
ahg-g: /ok-to-test
kerthcet: addressed the comments.

Patch:
diff --git a/PROJECT b/PROJECT
index bab428bcb..fe90a666b 100644
--- a/PROJECT
+++ b/PROJECT
@@ -38,11 +38,4 @@ resources:
   kind: ResourceFlavor
   path: sigs.k8s.io/kueue/api/v1alpha1
   version: v1alpha1
-- api:
-    crdVersion: v1
-  domain: x-k8s.io
-  group: config
-  kind: KueueConfiguration
-  path: sigs.k8s.io/kueue/apis/config/v1alpha1
-  version: v1alpha1
 version: "3"
diff --git a/README.md b/README.md
index 46d5efdc3..6ef6ab53a 100644
--- a/README.md
+++ b/README.md
@@ -24,10 +24,11 @@ presents the detailed design of the controller.
 
 **Requires Kubernetes 1.22 or newer**.
 
-To install the latest release of Kueue in your cluster, run the following command:
+To install the latest development version of Kueue in your cluster, run the
+following command:
 
 ```shell
-kubectl apply -f https://github.com/kubernetes-sigs/kueue/releases/download/v0.1.0/manifests.yaml
+kubectl apply -k github.com/kubernetes-sigs/kueue/config/default
 ```
 
 The controller runs in the `kueue-system` namespace.
diff --git a/apis/config/v1alpha1/configuration_types.go b/apis/config/v1alpha1/configuration_types.go
deleted file mode 100644
index 9db341e11..000000000
--- a/apis/config/v1alpha1/configuration_types.go
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
-Copyright 2022 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 v1alpha1
-
-import (
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	cfg "sigs.k8s.io/controller-runtime/pkg/config/v1alpha1"
-)
-
-//+kubebuilder:object:root=true
-
-// Configuration is the Schema for the kueueconfigurations API
-type Configuration struct {
-	metav1.TypeMeta `json:",inline"`
-
-	// ControllerManagerConfigurationSpec returns the configurations for controllers
-	cfg.ControllerManagerConfigurationSpec `json:",inline"`
-
-	// ManageJobsWithoutQueueName controls whether or not Kueue reconciles
-	// batch/v1.Jobs that don't set the annotation kueue.x-k8s.io/queue-name.
-	// If set to true, then those jobs will be suspended and never started unless
-	// they are assigned a queue and eventually admitted. This also applies to
-	// jobs created before starting the kueue controller.
-	// Defaults to false; therefore, those jobs are not managed and if they are created
-	// unsuspended, they will start immediately.
-	ManageJobsWithoutQueueName bool `json:"manageJobsWithoutQueueName"`
-}
-
-func init() {
-	SchemeBuilder.Register(&Configuration{})
-}
diff --git a/apis/config/v1alpha1/groupversion_info.go b/apis/config/v1alpha1/groupversion_info.go
deleted file mode 100644
index 37cc6cf82..000000000
--- a/apis/config/v1alpha1/groupversion_info.go
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
-Copyright 2022 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 v1alpha1 contains API Schema definitions for the config v1alpha1 API group
-//+kubebuilder:object:generate=true
-//+groupName=config.x-k8s.io
-package v1alpha1
-
-import (
-	"k8s.io/apimachinery/pkg/runtime/schema"
-	"sigs.k8s.io/controller-runtime/pkg/scheme"
-)
-
-var (
-	// GroupVersion is group version used to register these objects
-	GroupVersion = schema.GroupVersion{Group: "config.kueue.x-k8s.io", Version: "v1alpha1"}
-
-	// SchemeBuilder is used to add go types to the GroupVersionKind scheme
-	SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion}
-
-	// AddToScheme adds the types in this group-version to the given scheme.
-	AddToScheme = SchemeBuilder.AddToScheme
-)
diff --git a/apis/config/v1alpha1/zz_generated.deepcopy.go b/apis/config/v1alpha1/zz_generated.deepcopy.go
deleted file mode 100644
index 6c3786ea2..000000000
--- a/apis/config/v1alpha1/zz_generated.deepcopy.go
+++ /dev/null
@@ -1,51 +0,0 @@
-//go:build !ignore_autogenerated
-// +build !ignore_autogenerated
-
-/*
-Copyright 2022 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.
-*/
-
-// Code generated by controller-gen. DO NOT EDIT.
-
-package v1alpha1
-
-import (
-	runtime "k8s.io/apimachinery/pkg/runtime"
-)
-
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *Configuration) DeepCopyInto(out *Configuration) {
-	*out = *in
-	out.TypeMeta = in.TypeMeta
-	in.ControllerManagerConfigurationSpec.DeepCopyInto(&out.ControllerManagerConfigurationSpec)
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Configuration.
-func (in *Configuration) DeepCopy() *Configuration {
-	if in == nil {
-		return nil
-	}
-	out := new(Configuration)
-	in.DeepCopyInto(out)
-	return out
-}
-
-// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
-func (in *Configuration) DeepCopyObject() runtime.Object {
-	if c := in.DeepCopy(); c != nil {
-		return c
-	}
-	return nil
-}
diff --git a/config/default/kustomization.yaml b/config/default/kustomization.yaml
index 921044298..5f4a332df 100644
--- a/config/default/kustomization.yaml
+++ b/config/default/kustomization.yaml
@@ -32,7 +32,7 @@ patchesStrategicMerge:
 
 # Mount the controller config file for loading manager configurations
 # through a ComponentConfig type
-- manager_config_patch.yaml
+#- manager_config_patch.yaml
 
 # [WEBHOOK] To enable webhook, uncomment all the sections with [WEBHOOK] prefix including the one in
 # crd/kustomization.yaml
diff --git a/config/default/manager_auth_proxy_patch.yaml b/config/default/manager_auth_proxy_patch.yaml
index ca55a8117..15bb913b0 100644
--- a/config/default/manager_auth_proxy_patch.yaml
+++ b/config/default/manager_auth_proxy_patch.yaml
@@ -22,4 +22,7 @@ spec:
           name: https
       - name: manager
         args:
+        - "--health-probe-bind-address=:8081"
+        - "--metrics-bind-address=127.0.0.1:8080"
+        - "--leader-elect"
         - "--zap-log-level=2"
diff --git a/config/default/manager_config_patch.yaml b/config/default/manager_config_patch.yaml
index 7937e6666..6c400155c 100644
--- a/config/default/manager_config_patch.yaml
+++ b/config/default/manager_config_patch.yaml
@@ -10,7 +10,6 @@ spec:
       - name: manager
         args:
         - "--config=controller_manager_config.yaml"
-        - "--zap-log-level=2"
         volumeMounts:
         - name: manager-config
           mountPath: /controller_manager_config.yaml
diff --git a/config/manager/controller_manager_config.yaml b/config/manager/controller_manager_config.yaml
index 9a9df85b7..c3aa23da1 100644
--- a/config/manager/controller_manager_config.yaml
+++ b/config/manager/controller_manager_config.yaml
@@ -1,12 +1,11 @@
-apiVersion: config.kueue.x-k8s.io/v1alpha1
-kind: Configuration
+apiVersion: controller-runtime.sigs.k8s.io/v1alpha1
+kind: ControllerManagerConfig
 health:
   healthProbeBindAddress: :8081
 metrics:
-  bindAddress: :8080
+  bindAddress: 127.0.0.1:8080
 webhook:
   port: 9443
 leaderElection:
   leaderElect: true
   resourceName: c1f6bfd2.kueue.x-k8s.io
-#manageJobsWithoutQueueName: true
diff --git a/config/manager/kustomization.yaml b/config/manager/kustomization.yaml
index c166fcdcf..9ca115079 100644
--- a/config/manager/kustomization.yaml
+++ b/config/manager/kustomization.yaml
@@ -8,7 +8,6 @@ configMapGenerator:
 - files:
   - controller_manager_config.yaml
   name: manager-config
-
 apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 images:
diff --git a/docs/setup/install.md b/docs/setup/install.md
index 56570f953..0e9174a91 100644
--- a/docs/setup/install.md
+++ b/docs/setup/install.md
@@ -8,22 +8,13 @@ Make sure the following conditions are met:
 - The SuspendJob [feature gate](https://kubernetes.io/docs/reference/command-line-tools-reference/feature-gates/) is enabled. In Kubernetes 1.22 or newer, the feature gate is enabled by default.
 - The kubectl command-line tool has communication with your cluster.
 
-## Install a released version
-
-To install a released version of Kueue in your cluster, run the following command:
-
-```shell
-VERSION=v0.1.0
-kubectl apply -f https://github.com/kubernetes-sigs/kueue/releases/download/$VERSION/manifests.yaml
-```
-
 ## Install the latest development version
 
 To install the latest development version of Kueue in your cluster, run the
 following command:
 
 ```shell
-kubectl apply -k github.com/kubernetes-sigs/kueue/config/default?ref=main
+kubectl apply -k github.com/kubernetes-sigs/kueue/config/default
 ```
 
 The controller runs in the `kueue-system` namespace.
diff --git a/main.go b/main.go
index 508dfae21..4d23280aa 100644
--- a/main.go
+++ b/main.go
@@ -17,9 +17,7 @@ limitations under the License.
 package main
 
 import (
-	"bytes"
 	"flag"
-	"fmt"
 	"os"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
@@ -28,14 +26,12 @@ import (
 
 	schedulingv1 "k8s.io/api/scheduling/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/serializer"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
-	configv1alpha1 "sigs.k8s.io/kueue/apis/config/v1alpha1"
 	kueuev1alpha1 "sigs.k8s.io/kueue/apis/core/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/constants"
@@ -56,46 +52,32 @@ func init() {
 	utilruntime.Must(schedulingv1.AddToScheme(scheme))
 
 	utilruntime.Must(kueuev1alpha1.AddToScheme(scheme))
-	utilruntime.Must(configv1alpha1.AddToScheme(scheme))
 	//+kubebuilder:scaffold:scheme
 }
 
 func main() {
-	var configFile string
-	flag.StringVar(&configFile, "config", "",
-		"The controller will load its initial configuration from this file. "+
-			"Omit this flag to use the default configuration values. ")
-
+	var metricsAddr string
+	var enableLeaderElection bool
+	var probeAddr string
+	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
+	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
+	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
+		"Enable leader election for controller manager. "+
+			"Enabling this will ensure there is only one active controller manager.")
 	opts := zap.Options{}
 	opts.BindFlags(flag.CommandLine)
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
-	options := ctrl.Options{
+	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:                 scheme,
-		HealthProbeBindAddress: ":8081",
-		MetricsBindAddress:     ":8080",
+		MetricsBindAddress:     metricsAddr,
 		Port:                   9443,
+		HealthProbeBindAddress: probeAddr,
+		LeaderElection:         enableLeaderElection,
 		LeaderElectionID:       "c1f6bfd2.kueue.x-k8s.io",
-	}
-	var err error
-	config := configv1alpha1.Configuration{}
-	if configFile != "" {
-		options, err = options.AndFrom(ctrl.ConfigFile().AtPath(configFile).OfKind(&config))
-		if err != nil {
-			setupLog.Error(err, "unable to load the config file")
-			os.Exit(1)
-		}
-		cfgStr, err := encodeConfig(&config)
-		if err != nil {
-			setupLog.Error(err, "unable to encode config file")
-			os.Exit(1)
-		}
-		setupLog.Info("Successfully loaded config file", "config", cfgStr)
-	}
-
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), options)
+	})
 	if err != nil {
 		setupLog.Error(err, "unable to start manager")
 		os.Exit(1)
@@ -112,11 +94,8 @@ func main() {
 	if failedCtrl, err := core.SetupControllers(mgr, queues, cCache); err != nil {
 		setupLog.Error(err, "Unable to create controller", "controller", failedCtrl)
 	}
-	if err = job.NewReconciler(mgr.GetScheme(),
-		mgr.GetClient(),
-		mgr.GetEventRecorderFor(constants.JobControllerName),
-		job.WithManageJobsWithoutQueueName(config.ManageJobsWithoutQueueName),
-	).SetupWithManager(mgr); err != nil {
+	if err = job.NewReconciler(mgr.GetScheme(), mgr.GetClient(),
+		mgr.GetEventRecorderFor(constants.JobControllerName)).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Job")
 		os.Exit(1)
 	}
@@ -146,19 +125,3 @@ func main() {
 		os.Exit(1)
 	}
 }
-
-func encodeConfig(cfg *configv1alpha1.Configuration) (string, error) {
-	codecs := serializer.NewCodecFactory(scheme)
-	const mediaType = runtime.ContentTypeYAML
-	info, ok := runtime.SerializerInfoForMediaType(codecs.SupportedMediaTypes(), mediaType)
-	if !ok {
-		return "", fmt.Errorf("unable to locate encoder -- %q is not a supported media type", mediaType)
-	}
-
-	encoder := codecs.EncoderForVersion(info.Serializer, configv1alpha1.GroupVersion)
-	buf := new(bytes.Buffer)
-	if err := encoder.Encode(cfg, buf); err != nil {
-		return "", err
-	}
-	return buf.String(), nil
-}
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 097251282..8db5690bf 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -80,17 +80,22 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 
 	status := workloadStatus(&wl)
 	if status == pending && !r.queues.QueueForWorkloadExists(&wl) {
-		err := workload.UpdateWorkloadStatusIfChanged(ctx, r.client, &wl, kueue.WorkloadAdmitted, corev1.ConditionFalse,
+		err := workload.UpdateStatusIfChanged(ctx, r.client, &wl, kueue.WorkloadAdmitted, corev1.ConditionFalse,
 			"Inadmissible", fmt.Sprintf("Queue %s doesn't exist", wl.Spec.QueueName))
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 	cqName, cqOk := r.queues.ClusterQueueForWorkload(&wl)
 	if status == pending && !cqOk {
-		err := workload.UpdateWorkloadStatusIfChanged(ctx, r.client, &wl, kueue.WorkloadAdmitted, corev1.ConditionFalse,
+		err := workload.UpdateStatusIfChanged(ctx, r.client, &wl, kueue.WorkloadAdmitted, corev1.ConditionFalse,
 			"Inadmissible", fmt.Sprintf("ClusterQueue %s doesn't exist", cqName))
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
+	if status == admitted {
+		err := workload.UpdateStatusIfChanged(ctx, r.client, &wl, kueue.WorkloadAdmitted, corev1.ConditionTrue, "", "")
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+
 	return ctrl.Result{}, nil
 }
 
@@ -154,6 +159,7 @@ func (r *WorkloadReconciler) Update(e event.UpdateEvent) bool {
 	wl := e.ObjectNew.(*kueue.Workload)
 	defer r.notifyWatchers(oldWl)
 	defer r.notifyWatchers(wl)
+
 	status := workloadStatus(wl)
 	log := r.log.WithValues("workload", klog.KObj(wl), "queue", wl.Spec.QueueName, "status", status)
 	prevQueue := oldWl.Spec.QueueName
@@ -235,7 +241,7 @@ func (r *WorkloadReconciler) SetupWithManager(mgr ctrl.Manager) error {
 }
 
 func workloadStatus(w *kueue.Workload) string {
-	if workloadFinished(w) {
+	if workload.InCondition(w, kueue.WorkloadFinished) {
 		return finished
 	}
 	if w.Spec.Admission != nil {
@@ -243,12 +249,3 @@ func workloadStatus(w *kueue.Workload) string {
 	}
 	return pending
 }
-
-func workloadFinished(w *kueue.Workload) bool {
-	for _, c := range w.Status.Conditions {
-		if c.Type == kueue.WorkloadFinished {
-			return c.Status == corev1.ConditionTrue
-		}
-	}
-	return false
-}
diff --git a/pkg/controller/workload/job/job_controller.go b/pkg/controller/workload/job/job_controller.go
index 9798a4b2b..786a3da4f 100644
--- a/pkg/controller/workload/job/job_controller.go
+++ b/pkg/controller/workload/job/job_controller.go
@@ -46,24 +46,24 @@ var (
 
 // JobReconciler reconciles a Job object
 type JobReconciler struct {
-	client                     client.Client
-	scheme                     *runtime.Scheme
-	record                     record.EventRecorder
-	manageJobsWithoutQueueName bool
+	client                      client.Client
+	scheme                      *runtime.Scheme
+	record                      record.EventRecorder
+	processJobsWithoutQueueName bool
 }
 
 type options struct {
-	manageJobsWithoutQueueName bool
+	processJobsWithoutQueueName bool
 }
 
 // Option configures the reconciler.
 type Option func(*options)
 
-// WithManageJobsWithoutQueueName indicates if the controller should reconcile
+// WithProcessJobsWithoutQueueName indicates if the controller should reconcile
 // jobs that don't set the queue name annotation.
-func WithManageJobsWithoutQueueName(f bool) Option {
+func WithProcessJobsWithoutQueueName(f bool) Option {
 	return func(o *options) {
-		o.manageJobsWithoutQueueName = f
+		o.processJobsWithoutQueueName = f
 	}
 }
 
@@ -81,10 +81,10 @@ func NewReconciler(
 	}
 
 	return &JobReconciler{
-		scheme:                     scheme,
-		client:                     client,
-		record:                     record,
-		manageJobsWithoutQueueName: options.manageJobsWithoutQueueName,
+		scheme:                      scheme,
+		client:                      client,
+		record:                      record,
+		processJobsWithoutQueueName: options.processJobsWithoutQueueName,
 	}
 }
 
@@ -133,7 +133,7 @@ func (r *JobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 
 	log := ctrl.LoggerFrom(ctx).WithValues("job", klog.KObj(&job))
 	ctx = ctrl.LoggerInto(ctx, log)
-	if queueName(&job) == "" && !r.manageJobsWithoutQueueName {
+	if queueName(&job) == "" && !r.processJobsWithoutQueueName {
 		log.V(3).Info(fmt.Sprintf("%s annotation is not set, ignoring the job", constants.QueueAnnotation))
 		return ctrl.Result{}, nil
 	}
diff --git a/pkg/queue/cluster_queue_best_effort_fifo.go b/pkg/queue/cluster_queue_best_effort_fifo.go
index 98153cd4d..6e4e3de1c 100644
--- a/pkg/queue/cluster_queue_best_effort_fifo.go
+++ b/pkg/queue/cluster_queue_best_effort_fifo.go
@@ -75,15 +75,9 @@ func (cq *ClusterQueueBestEffortFIFO) Delete(w *kueue.Workload) {
 	cq.ClusterQueueImpl.Delete(w)
 }
 
-// RequeueIfNotPresent inserts a workload that cannot be admitted into
-// ClusterQueue, unless it is already in the queue. If immediate is true,
-// the workload will be pushed back to heap directly. If not,
-// the workload will be put into the inadmissibleWorkloads.
-func (cq *ClusterQueueBestEffortFIFO) RequeueIfNotPresent(wInfo *workload.Info, immediate bool) bool {
-	if immediate {
-		return cq.ClusterQueueImpl.pushIfNotPresent(wInfo)
-	}
-
+// AddInadmissibleIfNotPresent inserts a workload that cannot be admitted into
+// the inadmissibleWorkloads in ClusterQueue, unless it is already in the queue.
+func (cq *ClusterQueueBestEffortFIFO) AddInadmissibleIfNotPresent(wInfo *workload.Info) bool {
 	key := workload.Key(wInfo.Obj)
 	if cq.inadmissibleWorkloads[key] != nil {
 		return false
diff --git a/pkg/queue/cluster_queue_impl.go b/pkg/queue/cluster_queue_impl.go
index db9f2a49a..86117b1bb 100644
--- a/pkg/queue/cluster_queue_impl.go
+++ b/pkg/queue/cluster_queue_impl.go
@@ -92,7 +92,7 @@ func (c *ClusterQueueImpl) Delete(w *kueue.Workload) {
 	}
 }
 
-func (c *ClusterQueueImpl) RequeueIfNotPresent(wInfo *workload.Info, _ bool) bool {
+func (c *ClusterQueueImpl) AddInadmissibleIfNotPresent(wInfo *workload.Info) bool {
 	return c.pushIfNotPresent(wInfo)
 }
 
diff --git a/pkg/queue/cluster_queue_interface.go b/pkg/queue/cluster_queue_interface.go
index 52ca94b2e..0a2d760d4 100644
--- a/pkg/queue/cluster_queue_interface.go
+++ b/pkg/queue/cluster_queue_interface.go
@@ -49,16 +49,12 @@ type ClusterQueue interface {
 	// queue is empty.
 	Pop() *workload.Info
 
-	// RequeueIfNotPresent inserts a workload that was not
-	// admitted back into the ClusterQueue. If the boolean is true,
-	// the workloads should be put back in the queue immediately,
-	// because we couldn't determine if the workload was admissible
-	// in the last cycle. If the boolean is false, the implementation might
-	// choose to keep it in temporary placeholder stage where it doesn't
-	// compete with other workloads, until cluster events free up quota.
-	// The workload should not be reinserted if it's already in the ClusterQueue.
-	// Returns true if the workload was inserted.
-	RequeueIfNotPresent(*workload.Info, bool) bool
+	// AddInadmissibleIfNotPresent inserts a workload that could not be
+	// admitted back into the ClusterQueue. The implementation might choose to
+	// keep it in temporary placeholder stage where it doesn't compete with
+	// other workloads, until cluster events free up quota. The workload should
+	// not be reinserted if it's already in the ClusterQueue.
+	AddInadmissibleIfNotPresent(*workload.Info) bool
 	// QueueInadmissibleWorkloads moves all workloads put in temporary placeholder stage
 	// to the ClusterQueue. If at least one workload is moved,
 	// returns true. Otherwise returns false.
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index d91d5564a..701c65476 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -284,7 +284,7 @@ func (m *Manager) addOrUpdateWorkload(w *kueue.Workload) bool {
 // RequeueWorkload requeues the workload ensuring that the queue and the
 // workload still exist in the client cache and it's not admitted. It won't
 // requeue if the workload is already in the queue (possible if the workload was updated).
-func (m *Manager) RequeueWorkload(ctx context.Context, info *workload.Info, immediate bool) bool {
+func (m *Manager) RequeueWorkload(ctx context.Context, info *workload.Info) bool {
 	m.Lock()
 	defer m.Unlock()
 
@@ -306,7 +306,7 @@ func (m *Manager) RequeueWorkload(ctx context.Context, info *workload.Info, imme
 		return false
 	}
 
-	added := cq.RequeueIfNotPresent(info, immediate)
+	added := cq.AddInadmissibleIfNotPresent(info)
 	if added {
 		m.cond.Broadcast()
 	}
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 3683306e8..050ee93b0 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -109,6 +109,8 @@ func (s *Scheduler) schedule(ctx context.Context) {
 		}
 		c := snapshot.ClusterQueues[e.ClusterQueue]
 		if len(e.borrows) > 0 && c.Cohort != nil && usedCohorts.Has(c.Cohort.Name) {
+			// TODO(): we shouldn't emit an event or update the workload condition in this
+			// case, just re-queue directly.
 			e.status = skipped
 			e.inadmissibleReason = "cohort used in this cycle"
 			continue
@@ -117,7 +119,12 @@ func (s *Scheduler) schedule(ctx context.Context) {
 		if err := s.admit(ctrl.LoggerInto(ctx, log), e); err == nil {
 			e.status = assumed
 		} else {
+			log.Error(err, "Failed to admit workload")
 			e.inadmissibleReason = fmt.Sprintf("Failed to admit workload: %v", err)
+			err := workload.UpdateStatus(ctx, s.client, e.Obj, kueue.WorkloadAdmitted, corev1.ConditionFalse, "Pending", err.Error())
+			if err != nil {
+				log.Error(err, "Updating Workload status")
+			}
 		}
 		// Even if there was a failure, we shouldn't admit other workloads to this
 		// cohort.
@@ -134,7 +141,8 @@ func (s *Scheduler) schedule(ctx context.Context) {
 			"status", e.status,
 			"reason", e.inadmissibleReason)
 		if e.status != assumed {
-			s.requeueAndUpdate(log, ctx, e)
+			s.requeueAndUpdate(log, ctx, &e.Info, e.inadmissibleReason)
+			s.recorder.Eventf(e.Obj, corev1.EventTypeNormal, "Pending", e.inadmissibleReason)
 		}
 	}
 }
@@ -267,9 +275,9 @@ func (s *Scheduler) admit(ctx context.Context, e *entry) error {
 			log.V(2).Info("Workload not admitted because it was deleted")
 			return
 		}
-
 		log.Error(err, errCouldNotAdmitWL)
-		s.requeueAndUpdate(log, ctx, *e)
+		s.requeueAndUpdate(log, ctx, &e.Info, err.Error())
+		s.recorder.Eventf(newWorkload, corev1.EventTypeNormal, "Pending", "%s: %v", errCouldNotAdmitWL, err)
 	})
 
 	return nil
@@ -415,15 +423,11 @@ func (e entryOrdering) Less(i, j int) bool {
 	return a.Obj.CreationTimestamp.Before(&b.Obj.CreationTimestamp)
 }
 
-func (s *Scheduler) requeueAndUpdate(log logr.Logger, ctx context.Context, e entry) {
-	added := s.queues.RequeueWorkload(ctx, &e.Info, e.status != "")
-	log.V(2).Info("Workload re-queued", "workload", klog.KObj(e.Obj), "queue", klog.KRef(e.Obj.Namespace, e.Obj.Spec.QueueName), "added", added, "status", e.status)
-
-	if e.status == "" {
-		err := workload.UpdateStatus(ctx, s.client, e.Obj, kueue.WorkloadAdmitted, corev1.ConditionFalse, "Pending", e.inadmissibleReason)
-		if err != nil {
-			log.Error(err, "Could not update Workload status")
-		}
-		s.recorder.Eventf(e.Obj, corev1.EventTypeNormal, "Pending", e.inadmissibleReason)
+func (s *Scheduler) requeueAndUpdate(log logr.Logger, ctx context.Context, w *workload.Info, message string) {
+	added := s.queues.RequeueWorkload(ctx, w)
+	log.V(2).Info("Workload re-queued", "workload", klog.KObj(w.Obj), "queue", klog.KRef(w.Obj.Namespace, w.Obj.Spec.QueueName), "added", added)
+	err := workload.UpdateStatus(ctx, s.client, w.Obj, kueue.WorkloadAdmitted, corev1.ConditionFalse, "Pending", message)
+	if err != nil {
+		log.Error(err, "Could not update Workload status")
 	}
 }
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 4c459010d..d44ce16cc 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -162,10 +162,7 @@ func max(v1, v2 int64) int64 {
 // FindConditionIndex finds the provided condition from the given status and returns the index.
 // Returns -1 if the condition is not present.
 func FindConditionIndex(status *kueue.WorkloadStatus, conditionType kueue.WorkloadConditionType) int {
-	if status == nil {
-		return -1
-	}
-	if status.Conditions == nil {
+	if status == nil || status.Conditions == nil {
 		return -1
 	}
 	for i := range status.Conditions {
@@ -207,7 +204,7 @@ func UpdateStatus(ctx context.Context,
 	return c.Status().Update(ctx, &newWl)
 }
 
-func UpdateWorkloadStatusIfChanged(ctx context.Context,
+func UpdateStatusIfChanged(ctx context.Context,
 	c client.Client,
 	wl *kueue.Workload,
 	conditionType kueue.WorkloadConditionType,
@@ -226,3 +223,8 @@ func UpdateWorkloadStatusIfChanged(ctx context.Context,
 	// Updating an existing condition
 	return UpdateStatus(ctx, c, wl, conditionType, conditionStatus, reason, message)
 }
+
+func InCondition(w *kueue.Workload, condition kueue.WorkloadConditionType) bool {
+	i := FindConditionIndex(&w.Status, condition)
+	return i != -1 && w.Status.Conditions[i].Status == corev1.ConditionTrue
+}


Test patch:
diff --git a/pkg/queue/cluster_queue_best_effort_fifo_test.go b/pkg/queue/cluster_queue_best_effort_fifo_test.go
index fcd38e06e..6941810a5 100644
--- a/pkg/queue/cluster_queue_best_effort_fifo_test.go
+++ b/pkg/queue/cluster_queue_best_effort_fifo_test.go
@@ -100,7 +100,7 @@ func TestClusterQueueBestEffortFIFO(t *testing.T) {
 			}
 
 			for _, w := range test.inadmissibleWorkloadsToAdd {
-				cq.RequeueIfNotPresent(w, false)
+				cq.AddInadmissibleIfNotPresent(w)
 			}
 
 			for _, w := range test.workloadsToUpdate {
diff --git a/pkg/queue/manager_test.go b/pkg/queue/manager_test.go
index 5600e9190..bc678ac40 100644
--- a/pkg/queue/manager_test.go
+++ b/pkg/queue/manager_test.go
@@ -326,7 +326,7 @@ func TestStatus(t *testing.T) {
 	}
 }
 
-func TestRequeueWorkloadStrictFIFO(t *testing.T) {
+func TestRequeueWorkload(t *testing.T) {
 	scheme := runtime.NewScheme()
 	if err := kueue.AddToScheme(scheme); err != nil {
 		t.Fatalf("Failed adding kueue scheme: %s", err)
@@ -414,7 +414,7 @@ func TestRequeueWorkloadStrictFIFO(t *testing.T) {
 				_ = manager.AddOrUpdateWorkload(tc.workload)
 			}
 			info := workload.NewInfo(tc.workload)
-			if requeued := manager.RequeueWorkload(ctx, info, true); requeued != tc.wantRequeued {
+			if requeued := manager.RequeueWorkload(ctx, info); requeued != tc.wantRequeued {
 				t.Errorf("RequeueWorkload returned %t, want %t", requeued, tc.wantRequeued)
 			}
 		})
@@ -798,7 +798,7 @@ func TestHeadsAsync(t *testing.T) {
 				// Remove the initial workload from the manager.
 				mgr.Heads(ctx)
 				go func() {
-					mgr.RequeueWorkload(ctx, workload.NewInfo(&wl), true)
+					mgr.RequeueWorkload(ctx, workload.NewInfo(&wl))
 				}()
 			},
 		},
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 76022ee27..0442bdcf1 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -25,7 +25,6 @@ import (
 
 	logrtesting "github.com/go-logr/logr/testing"
 	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -33,7 +32,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/kueue/pkg/util/pointer"
 
@@ -1477,116 +1475,3 @@ func TestEntryOrdering(t *testing.T) {
 		t.Errorf("Unexpected order (-want,+got):\n%s", diff)
 	}
 }
-
-var ignoreConditionTimestamps = cmpopts.IgnoreFields(kueue.WorkloadCondition{}, "LastProbeTime", "LastTransitionTime")
-
-func TestRequeueAndUpdate(t *testing.T) {
-	cq := utiltesting.MakeClusterQueue("cq").QueueingStrategy(kueue.BestEffortFIFO).Obj()
-	q1 := utiltesting.MakeQueue("q1", "ns1").ClusterQueue("cq").Obj()
-	w1 := utiltesting.MakeWorkload("w1", "ns1").Queue("q1").Obj()
-
-	cases := []struct {
-		name          string
-		e             entry
-		wantWorkloads map[string]sets.String
-		wantStatus    kueue.WorkloadStatus
-	}{
-		{
-			name: "workload didn't fit",
-			e: entry{
-				status:             "",
-				inadmissibleReason: "didn't fit",
-			},
-			wantStatus: kueue.WorkloadStatus{
-				Conditions: []kueue.WorkloadCondition{
-					{
-						Type:    kueue.WorkloadAdmitted,
-						Status:  corev1.ConditionFalse,
-						Reason:  "Pending",
-						Message: "didn't fit",
-					},
-				},
-			},
-		},
-		{
-			name: "assumed",
-			e: entry{
-				status:             assumed,
-				inadmissibleReason: "",
-			},
-			wantWorkloads: map[string]sets.String{
-				"cq": sets.NewString(w1.Name),
-			},
-		},
-		{
-			name: "nominated",
-			e: entry{
-				status:             nominated,
-				inadmissibleReason: "failed to admit workload",
-			},
-			wantWorkloads: map[string]sets.String{
-				"cq": sets.NewString(w1.Name),
-			},
-		},
-		{
-			name: "skipped",
-			e: entry{
-				status:             skipped,
-				inadmissibleReason: "cohort used in this cycle",
-			},
-			wantWorkloads: map[string]sets.String{
-				"cq": sets.NewString(w1.Name),
-			},
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			log := logrtesting.NewTestLoggerWithOptions(t, logrtesting.Options{
-				Verbosity: 2,
-			})
-			ctx := ctrl.LoggerInto(context.Background(), log)
-			scheme := runtime.NewScheme()
-			if err := kueue.AddToScheme(scheme); err != nil {
-				t.Fatalf("Failed adding kueue scheme: %v", err)
-			}
-			if err := corev1.AddToScheme(scheme); err != nil {
-				t.Fatalf("Failed adding kueue scheme: %v", err)
-			}
-
-			clientBuilder := fake.NewClientBuilder().WithScheme(scheme).WithObjects(w1, q1, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ns1"}})
-			cl := clientBuilder.Build()
-			broadcaster := record.NewBroadcaster()
-			recorder := broadcaster.NewRecorder(scheme, corev1.EventSource{Component: constants.ManagerName})
-			qManager := queue.NewManager(cl)
-			cqCache := cache.New(cl)
-			scheduler := New(qManager, cqCache, cl, recorder)
-			if err := qManager.AddQueue(ctx, q1); err != nil {
-				t.Fatalf("Inserting queue %s/%s in manager: %v", q1.Namespace, q1.Name, err)
-			}
-			if err := qManager.AddClusterQueue(ctx, cq); err != nil {
-				t.Fatalf("Inserting clusterQueue %s in manager: %v", cq.Name, err)
-			}
-
-			wInfos := qManager.Heads(ctx)
-			if len(wInfos) != 1 {
-				t.Fatalf("Failed getting heads in cluster queue")
-			}
-			tc.e.Info = wInfos[0]
-			scheduler.requeueAndUpdate(log, ctx, tc.e)
-
-			qDump := qManager.Dump()
-			if diff := cmp.Diff(tc.wantWorkloads, qDump); diff != "" {
-				t.Errorf("Unexpected elements in the cluster queue (-want,+got):\n%s", diff)
-			}
-
-			var updatedWl kueue.Workload
-			if err := cl.Get(ctx, client.ObjectKeyFromObject(w1), &updatedWl); err != nil {
-				t.Fatalf("Failed obtaining updated object: %v", err)
-			}
-			if diff := cmp.Diff(tc.wantStatus, updatedWl.Status, ignoreConditionTimestamps); diff != "" {
-				t.Errorf("Unexpected status after updating (-want,+got):\n%s", diff)
-			}
-		})
-	}
-}
diff --git a/test/integration/controller/core/suite_test.go b/test/integration/controller/core/suite_test.go
index 01c662fc4..37da99586 100644
--- a/test/integration/controller/core/suite_test.go
+++ b/test/integration/controller/core/suite_test.go
@@ -61,7 +61,7 @@ var _ = ginkgo.AfterSuite(func() {
 	fwk.Teardown()
 })
 
-func managerSetup(mgr manager.Manager, ctx context.Context) {
+func managerSetup(mgr manager.Manager) {
 	err := queue.SetupIndexes(mgr.GetFieldIndexer())
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
diff --git a/test/integration/controller/core/workload_controller_test.go b/test/integration/controller/core/workload_controller_test.go
index 369de6190..291892df0 100644
--- a/test/integration/controller/core/workload_controller_test.go
+++ b/test/integration/controller/core/workload_controller_test.go
@@ -27,6 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	kueue "sigs.k8s.io/kueue/apis/core/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/util/testing"
+	"sigs.k8s.io/kueue/pkg/workload"
 	"sigs.k8s.io/kueue/test/integration/framework"
 )
 
@@ -39,6 +40,7 @@ var _ = ginkgo.Describe("Workload controller", func() {
 		queue                *kueue.Queue
 		wl                   *kueue.Workload
 		message              string
+		clusterQueue         *kueue.ClusterQueue
 	)
 
 	ginkgo.BeforeEach(func() {
@@ -105,4 +107,39 @@ var _ = ginkgo.Describe("Workload controller", func() {
 			gomega.Expect(updatedQueueWorkload.Status.Conditions[0].Message).To(testing.Equal(message))
 		})
 	})
+
+	ginkgo.When("the workload is admitted", func() {
+		ginkgo.BeforeEach(func() {
+			clusterQueue = testing.MakeClusterQueue("cluster-queue").
+				Resource(testing.MakeResource(resourceGPU).
+					Flavor(testing.MakeFlavor(flavorOnDemand, "5").Max("10").Obj()).Obj()).
+				Obj()
+			gomega.Expect(k8sClient.Create(ctx, clusterQueue)).To(gomega.Succeed())
+
+			queue = testing.MakeQueue("queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+			gomega.Expect(k8sClient.Create(ctx, queue)).To(gomega.Succeed())
+		})
+		ginkgo.AfterEach(func() {
+			gomega.Expect(framework.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			gomega.Expect(framework.DeleteClusterQueue(ctx, k8sClient, clusterQueue)).To(gomega.Succeed())
+			updatedQueueWorkload = kueue.Workload{}
+			clusterQueue = nil
+		})
+
+		ginkgo.It("Should update the workload's condition", func() {
+			ginkgo.By("Create workload")
+			wl = testing.MakeWorkload("one", ns.Name).Queue(queue.Name).Request(corev1.ResourceCPU, "1").Obj()
+			gomega.Expect(k8sClient.Create(ctx, wl)).To(gomega.Succeed())
+
+			ginkgo.By("Admit workload")
+			gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &updatedQueueWorkload)).To(gomega.Succeed())
+			updatedQueueWorkload.Spec.Admission = testing.MakeAdmission(clusterQueue.Name).
+				Flavor(corev1.ResourceCPU, flavorOnDemand).Obj()
+			gomega.Expect(k8sClient.Update(ctx, &updatedQueueWorkload)).To(gomega.Succeed())
+			gomega.Eventually(func() bool {
+				gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &updatedQueueWorkload)).To(gomega.Succeed())
+				return workload.InCondition(&updatedQueueWorkload, kueue.WorkloadAdmitted)
+			}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
+		})
+	})
 })
diff --git a/test/integration/controller/job/job_controller_test.go b/test/integration/controller/job/job_controller_test.go
index ca16e586d..c1ada5c20 100644
--- a/test/integration/controller/job/job_controller_test.go
+++ b/test/integration/controller/job/job_controller_test.go
@@ -64,7 +64,7 @@ var (
 var _ = ginkgo.Describe("Job controller", func() {
 	ginkgo.BeforeEach(func() {
 		fwk = &framework.Framework{
-			ManagerSetup: managerSetup(job.WithManageJobsWithoutQueueName(true)),
+			ManagerSetup: managerSetup(job.WithProcessJobsWithoutQueueName(true)),
 			CRDPath:      crdPath,
 		}
 		ctx, cfg, k8sClient = fwk.Setup()
diff --git a/test/integration/controller/job/suite_test.go b/test/integration/controller/job/suite_test.go
index 26ce40795..1a234ee59 100644
--- a/test/integration/controller/job/suite_test.go
+++ b/test/integration/controller/job/suite_test.go
@@ -17,7 +17,6 @@ limitations under the License.
 package job
 
 import (
-	"context"
 	"testing"
 
 	"github.com/onsi/ginkgo/v2"
@@ -39,7 +38,7 @@ func TestAPIs(t *testing.T) {
 }
 
 func managerSetup(opts ...job.Option) framework.ManagerSetup {
-	return func(mgr manager.Manager, ctx context.Context) {
+	return func(mgr manager.Manager) {
 		reconciler := job.NewReconciler(
 			mgr.GetScheme(),
 			mgr.GetClient(),
diff --git a/test/integration/framework/framework.go b/test/integration/framework/framework.go
index cece845f0..bc8ccbdcc 100644
--- a/test/integration/framework/framework.go
+++ b/test/integration/framework/framework.go
@@ -39,7 +39,7 @@ import (
 	//+kubebuilder:scaffold:imports
 )
 
-type ManagerSetup func(manager.Manager, context.Context)
+type ManagerSetup func(manager.Manager)
 
 type Framework struct {
 	CRDPath      string
@@ -76,9 +76,9 @@ func (f *Framework) Setup() (context.Context, *rest.Config, client.Client) {
 	})
 	gomega.ExpectWithOffset(1, err).NotTo(gomega.HaveOccurred(), "failed to create manager")
 
+	f.ManagerSetup(mgr)
 	ctx, cancel := context.WithCancel(context.Background())
 	f.cancel = cancel
-	f.ManagerSetup(mgr, ctx)
 
 	go func() {
 		defer ginkgo.GinkgoRecover()
diff --git a/test/integration/scheduler/scheduler_test.go b/test/integration/scheduler/scheduler_test.go
index f2dd99c77..5f1029095 100644
--- a/test/integration/scheduler/scheduler_test.go
+++ b/test/integration/scheduler/scheduler_test.go
@@ -103,7 +103,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		prodBEClusterQ = testing.MakeClusterQueue("prod-be-cq").
 			Cohort("be").QueueingStrategy(kueue.BestEffortFIFO).
 			Resource(testing.MakeResource(corev1.ResourceCPU).
-				Flavor(testing.MakeFlavor(onDemandFlavor.Name, "5").Max("15").Obj()).
+				Flavor(testing.MakeFlavor(onDemandFlavor.Name, "5").Max("10").Obj()).
 				Obj()).
 			Obj()
 		gomega.Expect(k8sClient.Create(ctx, prodBEClusterQ)).Should(gomega.Succeed())
@@ -111,7 +111,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		devBEClusterQ = testing.MakeClusterQueue("dev-be-cq").
 			Cohort("be").QueueingStrategy(kueue.BestEffortFIFO).
 			Resource(testing.MakeResource(corev1.ResourceCPU).
-				Flavor(testing.MakeFlavor(onDemandFlavor.Name, "5").Max("15").Obj()).
+				Flavor(testing.MakeFlavor(onDemandFlavor.Name, "5").Max("10").Obj()).
 				Obj()).
 			Obj()
 		gomega.Expect(k8sClient.Create(ctx, devBEClusterQ)).Should(gomega.Succeed())
@@ -147,10 +147,10 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		gomega.Expect(k8sClient.Create(ctx, prodJob1)).Should(gomega.Succeed())
 		lookupKey1 := types.NamespacedName{Name: prodJob1.Name, Namespace: prodJob1.Namespace}
 		createdProdJob1 := &batchv1.Job{}
-		gomega.Eventually(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, lookupKey1, createdProdJob1)).Should(gomega.Succeed())
-			return createdProdJob1.Spec.Suspend
-		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
+		gomega.Eventually(func() bool {
+			err := k8sClient.Get(ctx, lookupKey1, createdProdJob1)
+			return err == nil && !*createdProdJob1.Spec.Suspend
+		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(createdProdJob1.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
 
 		ginkgo.By("checking a second no-fit prod job does not start")
@@ -158,20 +158,18 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		gomega.Expect(k8sClient.Create(ctx, prodJob2)).Should(gomega.Succeed())
 		lookupKey2 := types.NamespacedName{Name: prodJob2.Name, Namespace: prodJob2.Namespace}
 		createdProdJob2 := &batchv1.Job{}
-		gomega.Consistently(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, lookupKey2, createdProdJob2)).Should(gomega.Succeed())
-			return createdProdJob2.Spec.Suspend
-		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		gomega.Consistently(func() bool {
+			return k8sClient.Get(ctx, lookupKey2, createdProdJob2) == nil && *createdProdJob2.Spec.Suspend
+		}, framework.ConsistentDuration, framework.Interval).Should(gomega.BeTrue())
 
 		ginkgo.By("checking a dev job starts")
 		devJob := testing.MakeJob("dev-job", ns.Name).Queue(devQueue.Name).Request(corev1.ResourceCPU, "5").Obj()
 		gomega.Expect(k8sClient.Create(ctx, devJob)).Should(gomega.Succeed())
 		createdDevJob := &batchv1.Job{}
-		gomega.Eventually(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: devJob.Name, Namespace: devJob.Namespace}, createdDevJob)).
-				Should(gomega.Succeed())
-			return createdDevJob.Spec.Suspend
-		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
+		gomega.Eventually(func() bool {
+			key := types.NamespacedName{Name: devJob.Name, Namespace: devJob.Namespace}
+			return k8sClient.Get(ctx, key, createdDevJob) == nil && !*createdDevJob.Spec.Suspend
+		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(createdDevJob.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(spotUntaintedFlavor.Name))
 
 		ginkgo.By("checking the second prod job starts when the first finishes")
@@ -183,10 +181,9 @@ var _ = ginkgo.Describe("Scheduler", func() {
 				LastTransitionTime: metav1.Now(),
 			})
 		gomega.Expect(k8sClient.Status().Update(ctx, createdProdJob1)).Should(gomega.Succeed())
-		gomega.Eventually(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, lookupKey2, createdProdJob2)).Should(gomega.Succeed())
-			return createdProdJob2.Spec.Suspend
-		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
+		gomega.Eventually(func() bool {
+			return k8sClient.Get(ctx, lookupKey2, createdProdJob2) == nil && !*createdProdJob2.Spec.Suspend
+		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(createdProdJob2.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
 	})
 
@@ -218,10 +215,10 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			Request(corev1.ResourceCPU, "5").Obj()
 		gomega.Expect(k8sClient.Create(ctx, job3)).Should(gomega.Succeed())
 		createdJob3 := &batchv1.Job{}
-		gomega.Eventually(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: job3.Name, Namespace: job3.Namespace}, createdJob3)).Should(gomega.Succeed())
-			return createdJob3.Spec.Suspend
-		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
+		gomega.Eventually(func() bool {
+			lookupKey := types.NamespacedName{Name: job3.Name, Namespace: job3.Namespace}
+			return k8sClient.Get(ctx, lookupKey, createdJob3) == nil && !*createdJob3.Spec.Suspend
+		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(createdJob3.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(spotTaintedFlavor.Name))
 	})
 
@@ -232,10 +229,9 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
 		lookupKey := types.NamespacedName{Name: job.Name, Namespace: job.Namespace}
 		createdJob := &batchv1.Job{}
-		gomega.Consistently(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob)).Should(gomega.Succeed())
-			return createdJob.Spec.Suspend
-		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		gomega.Consistently(func() bool {
+			return k8sClient.Get(ctx, lookupKey, createdJob) == nil && *createdJob.Spec.Suspend
+		}, framework.ConsistentDuration, framework.Interval).Should(gomega.BeTrue())
 
 		ginkgo.By("checking the job starts when a fallback ClusterQueue gets added")
 		fallbackClusterQueue := testing.MakeClusterQueue("fallback-cq").
@@ -249,10 +245,9 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		defer func() {
 			gomega.Expect(framework.DeleteClusterQueue(ctx, k8sClient, fallbackClusterQueue)).ToNot(gomega.HaveOccurred())
 		}()
-		gomega.Eventually(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob)).Should(gomega.Succeed())
-			return createdJob.Spec.Suspend
-		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
+		gomega.Eventually(func() bool {
+			return k8sClient.Get(ctx, lookupKey, createdJob) == nil && !*createdJob.Spec.Suspend
+		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(createdJob.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
 	})
 
@@ -338,19 +333,21 @@ var _ = ginkgo.Describe("Scheduler", func() {
 
 		ginkgo.By("checking that workload1 is created with priority and priorityName")
 		createdLowPriorityWorkload := &kueue.Workload{}
-		gomega.Eventually(func() error {
+		gomega.Eventually(func() bool {
 			lookupKey := types.NamespacedName{Name: jobLowPriority.Name, Namespace: jobLowPriority.Namespace}
-			return k8sClient.Get(ctx, lookupKey, createdLowPriorityWorkload)
-		}, framework.Timeout, framework.Interval).Should(gomega.Succeed())
+			err := k8sClient.Get(ctx, lookupKey, createdLowPriorityWorkload)
+			return err == nil
+		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(createdLowPriorityWorkload.Spec.PriorityClassName).Should(gomega.Equal(lowPriorityClass.Name))
 		gomega.Expect(*createdLowPriorityWorkload.Spec.Priority).Should(gomega.Equal(lowPriorityClass.Value))
 
 		ginkgo.By("checking that workload2 is created with priority and priorityName")
 		createdHighPriorityWorkload := &kueue.Workload{}
-		gomega.Eventually(func() error {
+		gomega.Eventually(func() bool {
 			lookupKey := types.NamespacedName{Name: jobHighPriority.Name, Namespace: jobHighPriority.Namespace}
-			return k8sClient.Get(ctx, lookupKey, createdHighPriorityWorkload)
-		}, framework.Timeout, framework.Interval).Should(gomega.Succeed())
+			err := k8sClient.Get(ctx, lookupKey, createdHighPriorityWorkload)
+			return err == nil
+		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(createdHighPriorityWorkload.Spec.PriorityClassName).Should(gomega.Equal(highPriorityClass.Name))
 		gomega.Expect(*createdHighPriorityWorkload.Spec.Priority).Should(gomega.Equal(highPriorityClass.Value))
 
@@ -359,11 +356,10 @@ var _ = ginkgo.Describe("Scheduler", func() {
 
 		ginkgo.By("checking the job with low priority continues to be suspended")
 		createdJob1 := &batchv1.Job{}
-		gomega.Consistently(func() *bool {
-			gomega.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: jobLowPriority.Name, Namespace: ns.Name},
-				createdJob1)).Should(gomega.Succeed())
-			return createdJob1.Spec.Suspend
-		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		gomega.Consistently(func() bool {
+			return k8sClient.Get(ctx, types.NamespacedName{Name: jobLowPriority.Name, Namespace: ns.Name},
+				createdJob1) == nil && *createdJob1.Spec.Suspend
+		}, framework.ConsistentDuration, framework.Interval).Should(gomega.BeTrue())
 
 		ginkgo.By("checking the job with high priority starts")
 		createdJob2 := &batchv1.Job{}
@@ -542,30 +538,4 @@ var _ = ginkgo.Describe("Scheduler", func() {
 
 		framework.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, prodClusterQ.Name, smallWl1, smallWl2)
 	})
-
-	ginkgo.It("Should schedule workloads borrowing quota from ClusterQueues in the same Cohort", func() {
-		wl1 := testing.MakeWorkload("wl-1", ns.Name).Queue(prodBEQueue.Name).Request(corev1.ResourceCPU, "11").Obj()
-		wl2 := testing.MakeWorkload("wl-2", ns.Name).Queue(devBEQueue.Name).Request(corev1.ResourceCPU, "11").Obj()
-
-		ginkgo.By("Creating two workloads")
-		gomega.Expect(k8sClient.Create(ctx, wl1)).Should(gomega.Succeed())
-		gomega.Expect(k8sClient.Create(ctx, wl2)).Should(gomega.Succeed())
-		framework.ExpectWorkloadsToBePending(ctx, k8sClient, wl1, wl2)
-
-		// Make sure workloads are in the same scheduling cycle.
-		testBEClusterQ := testing.MakeClusterQueue("test-be-cq").
-			Cohort("be").QueueingStrategy(kueue.BestEffortFIFO).
-			Resource(testing.MakeResource(corev1.ResourceCPU).
-				Flavor(testing.MakeFlavor(onDemandFlavor.Name,
-					"15").Max("15").Obj()).
-				Obj()).
-			Obj()
-		gomega.Expect(k8sClient.Create(ctx, testBEClusterQ)).Should(gomega.Succeed())
-		defer func() {
-			gomega.Expect(framework.DeleteClusterQueue(ctx, k8sClient, testBEClusterQ)).Should(gomega.Succeed())
-		}()
-
-		framework.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, prodBEClusterQ.Name, wl1)
-		framework.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, devBEClusterQ.Name, wl2)
-	})
 })
diff --git a/test/integration/scheduler/suite_test.go b/test/integration/scheduler/suite_test.go
index 2136461b9..a0db2b86f 100644
--- a/test/integration/scheduler/suite_test.go
+++ b/test/integration/scheduler/suite_test.go
@@ -64,7 +64,7 @@ var _ = ginkgo.AfterSuite(func() {
 	fwk.Teardown()
 })
 
-func managerAndSchedulerSetup(mgr manager.Manager, ctx context.Context) {
+func managerAndSchedulerSetup(mgr manager.Manager) {
 	err := queue.SetupIndexes(mgr.GetFieldIndexer())
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())