Repository: kubernetes-sigs/kueue

Issue:
Extend waitForPodsReady to evict a workload if a replacement pod cannot schedule 
**What would you like to be added**:

Extend the waitForPodsReady mechanism to allow workload eviction if there is a replacement pod which cannot be scheduled within a specified timeout.

**Why is this needed**:

Currently the mechanism only evicts workloads if they cannot start, but a pod can fail while the  workload is running.

**Completion requirements**:

This enhancement requires the following artifacts:

- [x] Design doc
- [x] API change
- [ ] Docs update

The artifacts should be linked in subsequent comments.


Hints from issue comments:
@mimowo: /assign @PBundyra @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@yaroslava-serdiuk: cc: @yaroslava-serdiuk @tenzen-y: /reopen

@PBundyra could you extend documentation in https://kueue.sigs.k8s.io/docs/tasks/manage/setup_wait_for_pods_ready/?
@k8s-ci-robot: @tenzen-y: Reopened this issue.

<details>

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

>/reopen
>
>@PBundyra could you extend documentation in https://kueue.sigs.k8s.io/docs/tasks/manage/setup_wait_for_pods_ready/?
>


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@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/@PBundyra: The field has been documented a while ago with this PR https://github.com/kubernetes-sigs/kueue/pull/4964
I believe we can close this issue as it was addressed
/close
@k8s-ci-robot: @PBundyra: Closing this issue.

<details>

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

>The field has been documented a while ago with this PR https://github.com/kubernetes-sigs/kueue/pull/4964
>I believe we can close this issue as it was addressed
>/close
>


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

Solution patch:
diff --git a/CHANGELOG/CHANGELOG-0.10.md b/CHANGELOG/CHANGELOG-0.10.md
index ddc4e6a0b..bf1a4e902 100644
--- a/CHANGELOG/CHANGELOG-0.10.md
+++ b/CHANGELOG/CHANGELOG-0.10.md
@@ -1,3 +1,40 @@
+## v0.10.2
+
+Changes since `v0.10.1`:
+
+## Changes by Kind
+
+### Bug or Regression
+
+- Add missing external types to apply configurations (#4201, @astefanutti)
+- Align default value for `managedJobsNamespaceSelector` in helm chart and kustomize files. (#4263, @dgrove-oss)
+- Disable the StatefulSet webhook in the kube-system and kueue-system namespaces by default.
+  This aligns the default StatefulSet webhook configuration with the Pod and Deployment configurations. (#4162, @dgrove-oss)
+- Fix a bug is incorrect field path in inadmissible reasons and messages when Pod resources requests do not satisfy LimitRange constraints. (#4291, @tenzen-y)
+- Fix a bug is incorrect field path in inadmissible reasons and messages when container requests exceed limits (#4245, @tenzen-y)
+- Fix a bug that allowed unsupported changes to some PodSpec fields which were resulting in the StatefulSet getting stuck on Pods with schedulingGates.
+
+  The validation blocks mutating the following Pod spec fields: `nodeSelector`, `affinity`, `tolerations`, `runtimeClassName`, `priority`, `topologySpreadConstraints`, `overhead`, `resourceClaims`, plus container (and init container) fields: `ports` and `resources.requests`.
+
+  Mutating other fields, such as container image, command or args, remains allowed and supported. (#4153, @mbobrovskyi)
+- Fix a bug that doesn't allow Kueue to delete Pods after a StatefulSet is deleted. (#4203, @mbobrovskyi)
+- Fix a bug that prevented tracking some of the controller-runtime metrics in Prometheus. (#4229, @tenzen-y)
+- Fix a bug truncating AdmissionCheck condition message at `1024` characters when creation of the associated ProvisioningRequest or PodTemplate fails.
+  Instead, use the `32*1024` characters limit as for condition messages. (#4194, @mbobrovskyi)
+- Fix the bug that prevented Kueue from updating the AdmissionCheck state in the Workload status on a ProvisioningRequest creation error. (#4117, @mbobrovskyi)
+- Helm: Fix the unspecified LeaderElection Role and Rolebinding namespaces (#4386, @eric-higgins-ai)
+- Improve error message in the event when scheduling for TAS workload fails due to unassigned flavors. (#4325, @mimowo)
+- MultiKueue: Do not update the status of the Job on the management cluster while the Job is suspended. This is updated  for jobs represented by JobSet, Kubeflow Jobs and MPIJob. (#4084, @IrvingMg)
+- Propagate the top-level setting of the `kueue.x-k8s.io/priority-class` label to the PodTemplate for
+  Deployments and StatefulSets. This way the Workload Priority class is no longer ignored by the workloads. (#4035, @Abirdcfly)
+- TAS: Fix a bug that incorrect topologies are assigned to Workloads when topology has insufficient allocatable Pods count (#4328, @tenzen-y)
+- TAS: Fix a bug that unschedulable nodes (".spec.unschedulable=true") are counted as allocatable capacities (#4205, @tenzen-y)
+- TAS: Fixed a bug that allows to create a JobSet with both kueue.x-k8s.io/podset-required-topology and kueue.x-k8s.io/podset-preferred-topology annotations set on the PodTemplate. (#4132, @mbobrovskyi)
+
+### Other (Cleanup or Flake)
+
+- Renamed Log key from "attemptCount" to "schedulingCycleCount". This key tracks how many scheduling cycles we have done since starting Kueue. (#4239, @gabesaba)
+
 ## v0.10.1
 
 Changes since `v0.10.0`:
diff --git a/CHANGELOG/CHANGELOG-0.9.md b/CHANGELOG/CHANGELOG-0.9.md
index 9776c6ffb..af0a81442 100644
--- a/CHANGELOG/CHANGELOG-0.9.md
+++ b/CHANGELOG/CHANGELOG-0.9.md
@@ -1,3 +1,37 @@
+## v0.9.4
+
+Changes since `v0.9.3`:
+
+## Changes by Kind
+
+### Bug or Regression
+
+- Add missing external types to apply configurations (#4202, @astefanutti)
+- Disable the StatefulSet webhook in the kube-system and kueue-system namespaces by default.
+  This aligns the default StatefulSet webhook configuration with the Pod and Deployment configurations. (#4161, @@dgrove-oss)
+- Fix a bug is incorrect field path in inadmissible reasons and messages when Pod resources requests do not satisfy LimitRange constraints. (#4290, @tenzen-y)
+- Fix a bug is incorrect field path in inadmissible reasons and messages when container requests exceed limits (#4246, @tenzen-y)
+- Fix a bug that allowed unsupported changes to some PodSpec fields which were resulting in the StatefulSet getting stuck on Pods with schedulingGates.
+
+  The validation blocks mutating the following Pod spec fields: `nodeSelector`, `affinity`, `tolerations`, `runtimeClassName`, `priority`, `topologySpreadConstraints`, `overhead`, `resourceClaims`, plus container (and init container) fields: `ports` and `resources.requests`.
+
+  Mutating other fields, such as container image, command or args, remains allowed and supported. (#4154, @mbobrovskyi)
+- Fix a bug that doesn't allow Kueue to delete Pods after a StatefulSet is deleted. (#4206, @mbobrovskyi)
+- Fix a bug that prevented tracking some of the controller-runtime metrics in Prometheus. (#4227, @tenzen-y)
+- Fix a bug truncating AdmissionCheck condition message at `1024` characters when creation of the associated ProvisioningRequest or PodTemplate fails.
+  Instead, use the `32*1024` characters limit as for condition messages. (#4195, @mbobrovskyi)
+- Fix the bug that prevented Kueue from updating the AdmissionCheck state in the Workload status on a ProvisioningRequest creation error. (#4118, @mbobrovskyi)
+- Helm: Fix the unspecified LeaderElection Role and Rolebinding namespaces (#4386, @eric-higgins-ai)
+- MultiKueue: Do not update the status of the Job on the management cluster while the Job is suspended. This is updated  for jobs represented by JobSet, Kubeflow Jobs and MPIJob. (#4085, @IrvingMg)
+- Propagate the top-level setting of the `kueue.x-k8s.io/priority-class` label to the PodTemplate for
+  Deployments and StatefulSets. This way the Workload Priority class is no longer ignored by the workloads. (#4036, @Abirdcfly)
+- TAS: Fix a bug that unschedulable nodes (".spec.unschedulable=true") are counted as allocatable capacities (#4209, @tenzen-y)
+- TAS: Fixed a bug that allows to create a JobSet with both kueue.x-k8s.io/podset-required-topology and kueue.x-k8s.io/podset-preferred-topology annotations set on the PodTemplate. (#4156, @mbobrovskyi)
+
+### Other (Cleanup or Flake)
+
+- Renamed Log key from "attemptCount" to "schedulingCycleCount". This key tracks how many scheduling cycles we have done since starting Kueue. (#4241, @tenzen-y)
+
 ## v0.9.3
 
 Changes since `v0.9.2`:
diff --git a/Makefile b/Makefile
index 1f45d8394..f949ffe0c 100644
--- a/Makefile
+++ b/Makefile
@@ -77,7 +77,7 @@ LD_FLAGS += -X '$(version_pkg).GitCommit=$(shell git rev-parse HEAD)'
 
 # Update these variables when preparing a new release or a release branch.
 # Then run `make prepare-release-branch`
-RELEASE_VERSION=v0.10.1
+RELEASE_VERSION=v0.10.2
 RELEASE_BRANCH=main
 
 .PHONY: all
diff --git a/README.md b/README.md
index 739daec5e..65c7aec60 100644
--- a/README.md
+++ b/README.md
@@ -58,7 +58,7 @@ Read the [overview](https://kueue.sigs.k8s.io/docs/overview/) and watch the Kueu
 To install the latest release of Kueue in your cluster, run the following command:
 
 ```shell
-kubectl apply --server-side -f https://github.com/kubernetes-sigs/kueue/releases/download/v0.10.1/manifests.yaml
+kubectl apply --server-side -f https://github.com/kubernetes-sigs/kueue/releases/download/v0.10.2/manifests.yaml
 ```
 
 The controller runs in the `kueue-system` namespace.
diff --git a/SECURITY-INSIGHTS.yaml b/SECURITY-INSIGHTS.yaml
index f7edc5875..f3a30d81b 100644
--- a/SECURITY-INSIGHTS.yaml
+++ b/SECURITY-INSIGHTS.yaml
@@ -1,11 +1,11 @@
 header:
   schema-version: 1.0.0
   expiration-date: '2024-09-28T01:00:00.000Z'
-  last-updated: '2025-01-20'
-  last-reviewed: '2025-01-20'
-  commit-hash: 404a0cd659b2331b232cf5ddf93a1b67d5cc146c
+  last-updated: '2025-02-25'
+  last-reviewed: '2025-02-25'
+  commit-hash: bc5b74090af4e54f702c41ffb6971170696f54b2
   project-url: 'https://github.com/kubernetes-sigs/kueue'
-  project-release: 0.10.1
+  project-release: 0.10.2
   changelog: 'https://github.com/kubernetes-sigs/kueue/tree/main/CHANGELOG'
   license: 'https://github.com/kubernetes-sigs/kueue/blob/main/LICENSE'
 project-lifecycle:
@@ -28,7 +28,7 @@ documentation:
   - 'https://kueue.sigs.k8s.io/docs/'
 distribution-points:
   - >-
-    https://github.com/kubernetes-sigs/kueue/releases/download/v0.10.1/manifests.yaml
+    https://github.com/kubernetes-sigs/kueue/releases/download/v0.10.2/manifests.yaml
 security-artifacts:
   threat-model:
     threat-model-created: false
@@ -62,5 +62,5 @@ dependencies:
   dependencies-lists:
     - 'https://github.com/kubernetes-sigs/kueue/blob/main/go.mod'
   sbom:
-    - sbom-file: https://github.com/kubernetes-sigs/kueue/releases/download/v0.10.1/kueue-v0.10.1.spdx.json
+    - sbom-file: https://github.com/kubernetes-sigs/kueue/releases/download/v0.10.2/kueue-v0.10.2.spdx.json
       sbom-format: SPDX
diff --git a/apis/kueue/v1beta1/clusterqueue_types.go b/apis/kueue/v1beta1/clusterqueue_types.go
index d0b385465..501555d85 100644
--- a/apis/kueue/v1beta1/clusterqueue_types.go
+++ b/apis/kueue/v1beta1/clusterqueue_types.go
@@ -110,7 +110,7 @@ type ClusterQueueSpec struct {
 	//   and there are admitted Workloads in the ClusterQueue with lower priority.
 	//
 	// The preemption algorithm tries to find a minimal set of Workloads to
-	// preempt to accomomdate the pending Workload, preempting Workloads with
+	// preempt to accommodate the pending Workload, preempting Workloads with
 	// lower priority first.
 	// +kubebuilder:default={}
 	Preemption *ClusterQueuePreemption `json:"preemption,omitempty"`
diff --git a/apis/kueue/v1beta1/multikueue_types.go b/apis/kueue/v1beta1/multikueue_types.go
index baa9506dc..be8f8363b 100644
--- a/apis/kueue/v1beta1/multikueue_types.go
+++ b/apis/kueue/v1beta1/multikueue_types.go
@@ -79,6 +79,8 @@ type MultiKueueClusterStatus struct {
 // +kubebuilder:subresource:status
 // +kubebuilder:resource:scope=Cluster
 
+// +kubebuilder:printcolumn:name="Connected",JSONPath=".status.conditions[?(@.type=='Active')].status",type="string",description="MultiKueueCluster is connected"
+// +kubebuilder:printcolumn:name="Age",JSONPath=".metadata.creationTimestamp",type="date",description="Time this workload was created"
 // MultiKueueCluster is the Schema for the multikueue API
 type MultiKueueCluster struct {
 	metav1.TypeMeta   `json:",inline"`
diff --git a/apis/kueue/v1beta1/resourceflavor_types.go b/apis/kueue/v1beta1/resourceflavor_types.go
index 944e9e73b..b7021b202 100644
--- a/apis/kueue/v1beta1/resourceflavor_types.go
+++ b/apis/kueue/v1beta1/resourceflavor_types.go
@@ -63,6 +63,8 @@ type ResourceFlavorSpec struct {
 	// have.
 	// Workloads' podsets must have tolerations for these nodeTaints in order to
 	// get assigned this ResourceFlavor during admission.
+	// Only the 'NoSchedule' and 'NoExecute' taint effects are evaluated,
+	// while 'PreferNoSchedule' is ignored.
 	//
 	// An example of a nodeTaint is
 	// cloud.provider.com/preemptible="true":NoSchedule
diff --git a/apis/kueue/v1beta1/workload_types.go b/apis/kueue/v1beta1/workload_types.go
index 88fcae5c2..15a8d1b0b 100644
--- a/apis/kueue/v1beta1/workload_types.go
+++ b/apis/kueue/v1beta1/workload_types.go
@@ -579,14 +579,20 @@ const (
 	// maximum execution time.
 	WorkloadMaximumExecutionTimeExceeded = "MaximumExecutionTimeExceeded"
 
-	// WorkloadWaitForPodsStart indicates the reason for PodsReady=False condition
+	// WorkloadWaitForStart indicates the reason for PodsReady=False condition
 	// when the pods have not been ready since admission, or the workload is not admitted.
-	WorkloadWaitForPodsStart = "WaitForPodsStart"
+	WorkloadWaitForStart = "WaitForStart"
 
-	// WorkloadWaitForPodsStart indicates the reason for the PodsReady=False condition
+	// WorkloadWaitForRecovery indicates the reason for the PodsReady=False condition
 	// when the Pods were ready since the workload admission, but some pod has failed,
 	// and workload waits for recovering.
-	WorkloadWaitForPodsRecovery = "WaitForPodsRecovery"
+	WorkloadWaitForRecovery = "WaitForRecovery"
+
+	// WorkloadStarted indicates that all Pods are ready and the Workload has successfully started
+	WorkloadStarted = "Started"
+
+	// WorkloadRecovered indicates that after at least one Pod has failed, the Workload has recovered and is running
+	WorkloadRecovered = "Recovered"
 )
 
 const (
diff --git a/charts/kueue/Chart.yaml b/charts/kueue/Chart.yaml
index f2a1fc9ae..da0a3e9c5 100644
--- a/charts/kueue/Chart.yaml
+++ b/charts/kueue/Chart.yaml
@@ -21,4 +21,4 @@ version: 0.1.0
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: "v0.10.1"
+appVersion: "v0.10.2"
diff --git a/charts/kueue/README.md b/charts/kueue/README.md
index 31acf6f2b..5178b97f3 100644
--- a/charts/kueue/README.md
+++ b/charts/kueue/README.md
@@ -3,12 +3,14 @@
 ## Table of Contents
 
 <!-- toc -->
-- [Installation](#installation)
-  - [Prerequisites](#prerequisites)
-  - [Installing the chart](#installing-the-chart)
-    - [Install chart using Helm v3.0+](#install-chart-using-helm-v30)
-    - [Verify that controller pods are running properly.](#verify-that-controller-pods-are-running-properly)
-  - [Configuration](#configuration)
+- [Kueue's helm chart](#kueues-helm-chart)
+  - [Table of Contents](#table-of-contents)
+    - [Installation](#installation)
+      - [Prerequisites](#prerequisites)
+      - [Installing the chart](#installing-the-chart)
+        - [Install chart using Helm v3.0+](#install-chart-using-helm-v30)
+        - [Verify that controller pods are running properly.](#verify-that-controller-pods-are-running-properly)
+    - [Configuration](#configuration)
 <!-- /toc -->
 
 ### Installation
@@ -35,7 +37,7 @@ $ helm install kueue kueue/ --create-namespace --namespace kueue-system
 Or use the charts pushed to `us-central1-docker.pkg.dev/k8s-staging-images/charts/kueue`:
 
 ```bash
-helm install kueue oci://us-central1-docker.pkg.dev/k8s-staging-images/charts/kueue --version="v0.10.1" --create-namespace --namespace=kueue-system
+helm install kueue oci://us-central1-docker.pkg.dev/k8s-staging-images/charts/kueue --version="v0.10.2" --create-namespace --namespace=kueue-system
 ```
 
 ##### Verify that controller pods are running properly.
@@ -57,6 +59,9 @@ The following table lists the configurable parameters of the kueue chart and the
 | `enablePrometheus`                                     | enable Prometheus                                      | `false`                                     |
 | `enableCertManager`                                    | enable CertManager                                     | `false`                                     |
 | `enableVisibilityAPF`                                  | enable APF for the visibility API                      | `false`                                     |
+| `enableKueueViz`                                       | enable KueueViz dashboard                              | `false`                                     |
+| `kueueViz.backend.image`                               | KueueViz dashboard backend image                       | `us-central1-docker.pkg.dev/k8s-staging-images/kueue/kueue-viz-backend:main-latest` |
+| `kueueViz.frontend.image`                              | KueueViz dashboard frontend image                      | `us-central1-docker.pkg.dev/k8s-staging-images/kueue/kueue-viz-frontend:main-latest` |
 | `controllerManager.manager.image.repository`           | controllerManager.manager's repository and image       | `us-central1-docker.pkg.dev/k8s-staging-images/kueue/kueue` |
 | `controllerManager.manager.image.tag`                  | controllerManager.manager's tag                        | `main`                                      |
 | `controllerManager.manager.resources`                  | controllerManager.manager's resources                  | abbr.                                       |
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
index 05074d616..4082bd40d 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
@@ -258,7 +258,7 @@ spec:
                     and there are admitted Workloads in the ClusterQueue with lower priority.
 
                   The preemption algorithm tries to find a minimal set of Workloads to
-                  preempt to accomomdate the pending Workload, preempting Workloads with
+                  preempt to accommodate the pending Workload, preempting Workloads with
                   lower priority first.
                 properties:
                   borrowWithinCohort:
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_multikueueclusters.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_multikueueclusters.yaml
index 81a611091..90dad4251 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_multikueueclusters.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_multikueueclusters.yaml
@@ -29,7 +29,16 @@ spec:
     singular: multikueuecluster
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: MultiKueueCluster is connected
+      jsonPath: .status.conditions[?(@.type=='Active')].status
+      name: Connected
+      type: string
+    - description: Time this workload was created
+      jsonPath: .metadata.creationTimestamp
+      name: Age
+      type: date
+    name: v1beta1
     schema:
       openAPIV3Schema:
         description: MultiKueueCluster is the Schema for the multikueue API
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_resourceflavors.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_resourceflavors.yaml
index b0df44d30..6f31405a3 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_resourceflavors.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_resourceflavors.yaml
@@ -81,6 +81,8 @@ spec:
                   have.
                   Workloads' podsets must have tolerations for these nodeTaints in order to
                   get assigned this ResourceFlavor during admission.
+                  Only the 'NoSchedule' and 'NoExecute' taint effects are evaluated,
+                  while 'PreferNoSchedule' is ignored.
 
                   An example of a nodeTaint is
                   cloud.provider.com/preemptible="true":NoSchedule
diff --git a/charts/kueue/templates/kueue-viz/backend-deployment.yaml b/charts/kueue/templates/kueue-viz/backend-deployment.yaml
new file mode 100644
index 000000000..aa6dbe76d
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/backend-deployment.yaml
@@ -0,0 +1,23 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-backend'
+  namespace: '{{ .Release.Namespace }}'
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: kueue-viz-backend
+  template:
+    metadata:
+      labels:
+        app: kueue-viz-backend
+    spec:
+      containers:
+        - name: backend
+          image: '{{ .Values.kueueViz.backend.image | default "us-central1-docker.pkg.dev/k8s-staging-images/kueue/kueue-viz-backend:main-latest" }}'
+          imagePullPolicy: IfNotPresent
+          ports:
+            - containerPort: 8080
+{{- end }}
diff --git a/charts/kueue/templates/kueue-viz/backend-ingress.yaml b/charts/kueue/templates/kueue-viz/backend-ingress.yaml
new file mode 100644
index 000000000..05703a5e1
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/backend-ingress.yaml
@@ -0,0 +1,26 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-backend-ingress'
+  namespace: '{{ .Release.Namespace }}'
+  annotations:
+    nginx.ingress.kubernetes.io/rewrite-target: /
+    nginx.ingress.kubernetes.io/ssl-redirect: "true"
+spec:
+  tls:
+    - hosts:
+        - backend.kueue-viz.local
+      secretName: '{{ include "kueue.fullname" . }}-kueue-viz-tls-secret'
+  rules:
+    - host: backend.kueue-viz.local
+      http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: '{{ include "kueue.fullname" . }}-kueue-viz-backend'
+                port:
+                  number: 8080
+{{- end }}
diff --git a/charts/kueue/templates/kueue-viz/backend-service.yaml b/charts/kueue/templates/kueue-viz/backend-service.yaml
new file mode 100644
index 000000000..4482aaa46
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/backend-service.yaml
@@ -0,0 +1,14 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: v1
+kind: Service
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-backend'
+  namespace: '{{ .Release.Namespace }}'
+spec:
+  type: ClusterIP
+  ports:
+    - port: 8080
+      targetPort: 8080
+  selector:
+    app: kueue-viz-backend
+{{- end }}
diff --git a/charts/kueue/templates/kueue-viz/cluster-role-binding.yaml b/charts/kueue/templates/kueue-viz/cluster-role-binding.yaml
new file mode 100644
index 000000000..89ef67fe3
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/cluster-role-binding.yaml
@@ -0,0 +1,15 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-backend-read-access-binding'
+  namespace: '{{ .Release.Namespace }}'
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-backend-read-access'
+subjects:
+  - kind: ServiceAccount
+    name: default
+    namespace: '{{ .Release.Namespace }}'
+{{- end }}
diff --git a/charts/kueue/templates/kueue-viz/clusterrole.yaml b/charts/kueue/templates/kueue-viz/clusterrole.yaml
new file mode 100644
index 000000000..a710f8fc4
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/clusterrole.yaml
@@ -0,0 +1,17 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-backend-read-access'
+  namespace: '{{ .Release.Namespace }}'
+rules:
+  - apiGroups: ["kueue.x-k8s.io"]
+    resources: ["workloads", "clusterqueues", "localqueues", "resourceflavors"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["pods", "events", "nodes"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: ["kueue.x-k8s.io"]
+    resources: ["workloadpriorityclass"]
+    verbs: ["get", "list", "watch"]
+{{- end }}
diff --git a/charts/kueue/templates/kueue-viz/frontend-deployment.yaml b/charts/kueue/templates/kueue-viz/frontend-deployment.yaml
new file mode 100644
index 000000000..395e5ae9d
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/frontend-deployment.yaml
@@ -0,0 +1,26 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-frontend'
+  namespace: '{{ .Release.Namespace }}'
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: kueue-viz-frontend
+  template:
+    metadata:
+      labels:
+        app: kueue-viz-frontend
+    spec:
+      containers:
+        - name: frontend
+          image: '{{ .Values.kueueViz.frontend.image | default "us-central1-docker.pkg.dev/k8s-staging-images/kueue/kueue-viz-frontend:main-latest" }}'
+          imagePullPolicy: IfNotPresent
+          ports:
+            - containerPort: 8080
+          env:
+            - name: REACT_APP_WEBSOCKET_URL
+              value: "wss://backend.kueue-viz.local"
+{{- end }}
diff --git a/charts/kueue/templates/kueue-viz/frontend-ingress.yaml b/charts/kueue/templates/kueue-viz/frontend-ingress.yaml
new file mode 100644
index 000000000..ddf97f900
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/frontend-ingress.yaml
@@ -0,0 +1,26 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-frontend-ingress'
+  namespace: '{{ .Release.Namespace }}'
+  annotations:
+    nginx.ingress.kubernetes.io/rewrite-target: /
+    nginx.ingress.kubernetes.io/ssl-redirect: "true"
+spec:
+  tls:
+    - hosts:
+        - frontend.kueue-viz.local
+      secretName: '{{ include "kueue.fullname" . }}-kueue-viz-tls-secret'
+  rules:
+    - host: frontend.kueue-viz.local
+      http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: '{{ include "kueue.fullname" . }}-kueue-viz-frontend'
+                port:
+                  number: 8080
+{{- end }}
diff --git a/charts/kueue/templates/kueue-viz/frontend-service.yaml b/charts/kueue/templates/kueue-viz/frontend-service.yaml
new file mode 100644
index 000000000..3bfb9946a
--- /dev/null
+++ b/charts/kueue/templates/kueue-viz/frontend-service.yaml
@@ -0,0 +1,14 @@
+{{- if .Values.enableKueueViz }}
+apiVersion: v1
+kind: Service
+metadata:
+  name: '{{ include "kueue.fullname" . }}-kueue-viz-frontend'
+  namespace: '{{ .Release.Namespace }}'
+spec:
+  type: ClusterIP
+  ports:
+    - port: 8080
+      targetPort: 8080
+  selector:
+    app: kueue-viz-frontend
+{{- end }}
diff --git a/charts/kueue/values.yaml b/charts/kueue/values.yaml
index 41860f387..1111b7a9e 100644
--- a/charts/kueue/values.yaml
+++ b/charts/kueue/values.yaml
@@ -150,3 +150,6 @@ webhookService:
       protocol: TCP
       targetPort: 9443
   type: ClusterIP
+
+# kueue-viz dashboard
+enableKueueViz: false
diff --git a/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml b/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
index 9e7b8b2f9..c7f008943 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
@@ -243,7 +243,7 @@ spec:
                     and there are admitted Workloads in the ClusterQueue with lower priority.
 
                   The preemption algorithm tries to find a minimal set of Workloads to
-                  preempt to accomomdate the pending Workload, preempting Workloads with
+                  preempt to accommodate the pending Workload, preempting Workloads with
                   lower priority first.
                 properties:
                   borrowWithinCohort:
diff --git a/config/components/crd/bases/kueue.x-k8s.io_multikueueclusters.yaml b/config/components/crd/bases/kueue.x-k8s.io_multikueueclusters.yaml
index 6f5cf7718..1d334815b 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_multikueueclusters.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_multikueueclusters.yaml
@@ -14,7 +14,16 @@ spec:
     singular: multikueuecluster
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: MultiKueueCluster is connected
+      jsonPath: .status.conditions[?(@.type=='Active')].status
+      name: Connected
+      type: string
+    - description: Time this workload was created
+      jsonPath: .metadata.creationTimestamp
+      name: Age
+      type: date
+    name: v1beta1
     schema:
       openAPIV3Schema:
         description: MultiKueueCluster is the Schema for the multikueue API
diff --git a/config/components/crd/bases/kueue.x-k8s.io_resourceflavors.yaml b/config/components/crd/bases/kueue.x-k8s.io_resourceflavors.yaml
index acd26acca..a66fb1df9 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_resourceflavors.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_resourceflavors.yaml
@@ -66,6 +66,8 @@ spec:
                   have.
                   Workloads' podsets must have tolerations for these nodeTaints in order to
                   get assigned this ResourceFlavor during admission.
+                  Only the 'NoSchedule' and 'NoExecute' taint effects are evaluated,
+                  while 'PreferNoSchedule' is ignored.
 
                   An example of a nodeTaint is
                   cloud.provider.com/preemptible="true":NoSchedule
diff --git a/config/components/kueue-viz/backend-deployment.yaml b/config/components/kueue-viz/backend-deployment.yaml
new file mode 100644
index 000000000..7782a9c7b
--- /dev/null
+++ b/config/components/kueue-viz/backend-deployment.yaml
@@ -0,0 +1,22 @@
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: kueue-viz-backend
+  namespace: system
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: kueue-viz-backend
+  template:
+    metadata:
+      labels:
+        app: kueue-viz-backend
+    spec:
+      containers:
+        - name: backend
+          image: us-central1-docker.pkg.dev/k8s-staging-images/kueue/kueue-viz-backend:main-latest
+          imagePullPolicy: IfNotPresent
+          ports:
+            - containerPort: 8080
diff --git a/config/components/kueue-viz/backend-ingress.yaml b/config/components/kueue-viz/backend-ingress.yaml
new file mode 100644
index 000000000..573201293
--- /dev/null
+++ b/config/components/kueue-viz/backend-ingress.yaml
@@ -0,0 +1,26 @@
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: kueue-viz-backend-ingress
+  namespace: system
+  annotations:
+    nginx.ingress.kubernetes.io/rewrite-target: /
+    nginx.ingress.kubernetes.io/ssl-redirect: "true"
+spec:
+  tls:
+    - hosts:
+        - backend.kueue-viz.local
+      secretName: kueue-viz-tls-secret
+  rules:
+    - host: backend.kueue-viz.local
+      http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: kueue-viz-backend
+                port:
+                  number: 8080
+
diff --git a/config/components/kueue-viz/backend-service.yaml b/config/components/kueue-viz/backend-service.yaml
new file mode 100644
index 000000000..a34c2dbcb
--- /dev/null
+++ b/config/components/kueue-viz/backend-service.yaml
@@ -0,0 +1,13 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: kueue-viz-backend
+  namespace: system
+spec:
+  type: ClusterIP
+  ports:
+    - port: 8080
+      targetPort: 8080
+  selector:
+    app: kueue-viz-backend
diff --git a/config/components/kueue-viz/cluster-role-binding.yaml b/config/components/kueue-viz/cluster-role-binding.yaml
new file mode 100644
index 000000000..b3d38aa59
--- /dev/null
+++ b/config/components/kueue-viz/cluster-role-binding.yaml
@@ -0,0 +1,13 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: kueue-viz-backend-read-access-binding
+  namespace: system
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: kueue-viz-backend-read-access
+subjects:
+  - kind: ServiceAccount
+    name: default
+    namespace: system
diff --git a/config/components/kueue-viz/clusterrole.yaml b/config/components/kueue-viz/clusterrole.yaml
new file mode 100644
index 000000000..79eba67fb
--- /dev/null
+++ b/config/components/kueue-viz/clusterrole.yaml
@@ -0,0 +1,15 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: kueue-viz-backend-read-access
+  namespace: system
+rules:
+  - apiGroups: ["kueue.x-k8s.io"]
+    resources: ["workloads", "clusterqueues", "localqueues", "resourceflavors"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["pods", "events", "nodes"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: ["kueue.x-k8s.io"]
+    resources: ["workloadpriorityclass"]
+    verbs: ["get", "list", "watch"]
diff --git a/config/components/kueue-viz/frontend-deployment.yaml b/config/components/kueue-viz/frontend-deployment.yaml
new file mode 100644
index 000000000..909a6c9a1
--- /dev/null
+++ b/config/components/kueue-viz/frontend-deployment.yaml
@@ -0,0 +1,26 @@
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: kueue-viz-frontend
+  namespace: system
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: kueue-viz-frontend
+  template:
+    metadata:
+      labels:
+        app: kueue-viz-frontend
+    spec:
+
+      containers:
+        - name: frontend
+          image: us-central1-docker.pkg.dev/k8s-staging-images/kueue/kueue-viz-frontend:main-latest
+          imagePullPolicy: IfNotPresent
+          ports:
+            - containerPort: 8080
+          env:
+            - name: REACT_APP_WEBSOCKET_URL
+              value: "wss://backend.kueue-viz.local"
diff --git a/config/components/kueue-viz/frontend-ingress.yaml b/config/components/kueue-viz/frontend-ingress.yaml
new file mode 100644
index 000000000..eb3a3700f
--- /dev/null
+++ b/config/components/kueue-viz/frontend-ingress.yaml
@@ -0,0 +1,25 @@
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: kueue-viz-frontend-ingress
+  namespace: system
+  annotations:
+    nginx.ingress.kubernetes.io/rewrite-target: /
+    nginx.ingress.kubernetes.io/ssl-redirect: "true"
+spec:
+  tls:
+    - hosts:
+        - frontend.kueue-viz.local
+      secretName: kueue-viz-tls-secret
+  rules:
+    - host: frontend.kueue-viz.local
+      http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: kueue-viz-frontend
+                port:
+                  number: 8080
diff --git a/config/components/kueue-viz/frontend-service.yaml b/config/components/kueue-viz/frontend-service.yaml
new file mode 100644
index 000000000..9a931555d
--- /dev/null
+++ b/config/components/kueue-viz/frontend-service.yaml
@@ -0,0 +1,13 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: kueue-viz-frontend
+  namespace: system
+spec:
+  type: ClusterIP
+  ports:
+    - port: 8080
+      targetPort: 8080
+  selector:
+    app: kueue-viz-frontend
diff --git a/config/components/kueue-viz/kustomization.yaml b/config/components/kueue-viz/kustomization.yaml
new file mode 100644
index 000000000..30ca19740
--- /dev/null
+++ b/config/components/kueue-viz/kustomization.yaml
@@ -0,0 +1,9 @@
+resources:
+  - clusterrole.yaml
+  - cluster-role-binding.yaml
+  - backend-deployment.yaml
+  - frontend-deployment.yaml
+  - backend-service.yaml
+  - frontend-service.yaml
+  - backend-ingress.yaml
+  - frontend-ingress.yaml
diff --git a/config/kueue-viz/kustomization.yaml b/config/kueue-viz/kustomization.yaml
new file mode 100644
index 000000000..4a71c856b
--- /dev/null
+++ b/config/kueue-viz/kustomization.yaml
@@ -0,0 +1,13 @@
+# This overlay builds the kueue-viz dashboard to be used in combination
+# with other overlays.
+
+namespace: kueue-system
+namePrefix: kueue-
+
+labels:
+  - pairs:
+      app.kubernetes.io/name: kueue
+      app.kubernetes.io/component: dashboard
+
+resources:
+  - ../components/kueue-viz/
diff --git a/hack/update-helm.sh b/hack/update-helm.sh
index 1e4874ce0..84edb4964 100755
--- a/hack/update-helm.sh
+++ b/hack/update-helm.sh
@@ -24,12 +24,14 @@ SRC_RBAC_DIR=config/components/rbac
 SRC_WEBHOOK_DIR=config/components/webhook
 SRC_VISIBILITY_DIR=config/components/visibility
 SRC_VISIBILITY_APF_DIR=config/components/visibility-apf
+SRC_KUEUE_VIZ_DIR=config/components/kueue-viz
 
 DEST_CRD_DIR=charts/kueue/templates/crd
 DEST_RBAC_DIR=charts/kueue/templates/rbac
 DEST_WEBHOOK_DIR=charts/kueue/templates/webhook
 DEST_VISIBILITY_DIR=charts/kueue/templates/visibility
 DEST_VISIBILITY_APF_DIR=charts/kueue/templates/visibility-apf
+DEST_KUEUE_VIZ_DIR=charts/kueue/templates/kueue-viz
 
 YQ=./bin/yq
 SED=${SED:-/usr/bin/sed}
@@ -52,6 +54,8 @@ find $SRC_WEBHOOK_DIR -name "*.yaml" $EXCLUDE_FILES_ARGS -exec cp "{}" $DEST_WEB
 find $SRC_VISIBILITY_DIR -name "*.yaml" $EXCLUDE_FILES_ARGS -exec cp "{}" $DEST_VISIBILITY_DIR \;
 # shellcheck disable=SC2086
 find $SRC_VISIBILITY_APF_DIR -name "*.yaml" $EXCLUDE_FILES_ARGS -exec cp "{}" $DEST_VISIBILITY_APF_DIR \;
+# shellcheck disable=SC2086
+find $SRC_KUEUE_VIZ_DIR -name "*.yaml" $EXCLUDE_FILES_ARGS -exec cp "{}" $DEST_KUEUE_VIZ_DIR \;
 $YQ -N -s '.kind' ${DEST_WEBHOOK_DIR}/manifests.yaml
 rm ${DEST_WEBHOOK_DIR}/manifests.yaml
 files=("MutatingWebhookConfiguration.yml" "ValidatingWebhookConfiguration.yml")
@@ -421,3 +425,46 @@ for output_file in "${DEST_VISIBILITY_APF_DIR}"/*.yaml; do
   } > "${output_file}.tmp"
   mv "${output_file}.tmp" "${output_file}"
 done
+
+# Add kueue-viz templating on kueue-viz directory
+for output_file in "${DEST_KUEUE_VIZ_DIR}"/*.yaml; do
+  if [ "$(< "$output_file" $YQ '.kind | select(. == "Deployment")')" ]; then
+    # Add Helm parameters for backend and frontend images with default values
+    deployment_name="$($YQ '.metadata.name' "$output_file")"
+if [ "$deployment_name" = "kueue-viz-backend" ]; then
+      original_image="$($YQ '.spec.template.spec.containers[0].image' "$output_file")"
+      $YQ -N -i \
+        ".spec.template.spec.containers[0].image = \"{{ .Values.kueueViz.backend.image | default \\\"$original_image\\\" }}\"" \
+        "$output_file"
+    fi
+    if [ "$deployment_name" = "kueue-viz-frontend" ]; then
+      original_image="$($YQ '.spec.template.spec.containers[0].image' "$output_file")"
+      $YQ -N -i \
+        ".spec.template.spec.containers[0].image = \"{{ .Values.kueueViz.frontend.image | default \\\"$original_image\\\" }}\"" \
+        "$output_file"
+    fi
+  fi
+
+  $YQ -N -i '.metadata.name |= "{{ include \"kueue.fullname\" . }}-" + .' "$output_file"
+
+  if [ "$(< "$output_file" $YQ '.kind | select(. == "Ingress")')" ]; then
+    $YQ -N -i '.spec.rules.[].http.paths.[].backend.service.name |= "{{ include \"kueue.fullname\" . }}-" + .' "$output_file"
+    $YQ -N -i '.spec.tls.[].secretName |= "{{ include \"kueue.fullname\" . }}-" + .' "$output_file"
+  fi
+
+  if [ "$(< "$output_file" $YQ '.kind | select(. == "ClusterRoleBinding")')" ]; then
+    $YQ -N -i '.roleRef.name |= "{{ include \"kueue.fullname\" . }}-" + .' "$output_file"
+  fi
+
+  if [ "$(< "$output_file" $YQ '.subjects.[] | has("namespace")')" = "true" ]; then
+    $YQ -N -i '.subjects.[].namespace = "{{ .Release.Namespace }}"' "$output_file"
+  fi
+  $YQ -N -i '.metadata.namespace = "{{ .Release.Namespace }}"' "$output_file"
+  {
+    echo '{{- if .Values.enableKueueViz }}'
+    cat "$output_file"
+    echo "{{- end }}"
+  } > "${output_file}.tmp"
+  mv "${output_file}.tmp" "${output_file}"
+done
+
diff --git a/pkg/controller/admissionchecks/multikueue/admissioncheck.go b/pkg/controller/admissionchecks/multikueue/admissioncheck.go
index 0412ea560..5ef7a0956 100644
--- a/pkg/controller/admissionchecks/multikueue/admissioncheck.go
+++ b/pkg/controller/admissionchecks/multikueue/admissioncheck.go
@@ -62,12 +62,12 @@ type ACReconciler struct {
 var _ reconcile.Reconciler = (*ACReconciler)(nil)
 
 func (a *ACReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
-	log := ctrl.LoggerFrom(ctx)
 	ac := &kueue.AdmissionCheck{}
 	if err := a.client.Get(ctx, req.NamespacedName, ac); err != nil || ac.Spec.ControllerName != kueue.MultiKueueControllerName {
 		return reconcile.Result{}, client.IgnoreNotFound(err)
 	}
 
+	log := ctrl.LoggerFrom(ctx)
 	log.V(2).Info("Reconcile AdmissionCheck")
 
 	newCondition := metav1.Condition{
diff --git a/pkg/controller/admissionchecks/multikueue/multikueuecluster.go b/pkg/controller/admissionchecks/multikueue/multikueuecluster.go
index 62e284e40..b0fb15e76 100644
--- a/pkg/controller/admissionchecks/multikueue/multikueuecluster.go
+++ b/pkg/controller/admissionchecks/multikueue/multikueuecluster.go
@@ -389,13 +389,13 @@ func (c *clustersReconciler) controllerFor(acName string) (*remoteClient, bool)
 
 func (c *clustersReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	cluster := &kueue.MultiKueueCluster{}
-	log := ctrl.LoggerFrom(ctx)
 
 	err := c.localClient.Get(ctx, req.NamespacedName, cluster)
 	if client.IgnoreNotFound(err) != nil {
 		return reconcile.Result{}, err
 	}
 
+	log := ctrl.LoggerFrom(ctx)
 	log.V(2).Info("Reconcile MultiKueueCluster")
 
 	if err != nil || !cluster.DeletionTimestamp.IsZero() {
diff --git a/pkg/controller/admissionchecks/multikueue/workload.go b/pkg/controller/admissionchecks/multikueue/workload.go
index 4d5b72eac..762b9572a 100644
--- a/pkg/controller/admissionchecks/multikueue/workload.go
+++ b/pkg/controller/admissionchecks/multikueue/workload.go
@@ -156,6 +156,7 @@ func (g *wlGroup) RemoveRemoteObjects(ctx context.Context, cluster string) error
 func (w *wlReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	log := ctrl.LoggerFrom(ctx)
 	log.V(2).Info("Reconcile Workload")
+
 	wl := &kueue.Workload{}
 	isDeleted := false
 	err := w.client.Get(ctx, req.NamespacedName, wl)
diff --git a/pkg/controller/admissionchecks/provisioning/controller.go b/pkg/controller/admissionchecks/provisioning/controller.go
index 949967353..6dc6c41e3 100644
--- a/pkg/controller/admissionchecks/provisioning/controller.go
+++ b/pkg/controller/admissionchecks/provisioning/controller.go
@@ -115,11 +115,15 @@ func NewController(client client.Client, record record.EventRecorder) (*Controll
 // Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
 func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	wl := &kueue.Workload{}
-	log := ctrl.LoggerFrom(ctx)
+
 	err := c.client.Get(ctx, req.NamespacedName, wl)
 	if err != nil {
 		return reconcile.Result{}, client.IgnoreNotFound(err)
 	}
+
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile Workload")
+
 	if !workload.HasQuotaReservation(wl) || workload.IsFinished(wl) || workload.IsEvicted(wl) {
 		return reconcile.Result{}, nil
 	}
diff --git a/pkg/controller/core/admissioncheck_controller.go b/pkg/controller/core/admissioncheck_controller.go
index e3a268e06..31880cb5f 100644
--- a/pkg/controller/core/admissioncheck_controller.go
+++ b/pkg/controller/core/admissioncheck_controller.go
@@ -81,7 +81,9 @@ func (r *AdmissionCheckReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	log := log.FromContext(ctx).WithValues("admissionCheck", klog.KObj(ac))
+	log := log.FromContext(ctx)
+	log.V(2).Info("Reconcile AdmissionCheck")
+
 	if ac.DeletionTimestamp.IsZero() {
 		if controllerutil.AddFinalizer(ac, kueue.ResourceInUseFinalizerName) {
 			if err := r.client.Update(ctx, ac); err != nil {
diff --git a/pkg/controller/core/clusterqueue_controller.go b/pkg/controller/core/clusterqueue_controller.go
index 3ebc421d9..700501bec 100644
--- a/pkg/controller/core/clusterqueue_controller.go
+++ b/pkg/controller/core/clusterqueue_controller.go
@@ -171,9 +171,9 @@ func (r *ClusterQueueReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		// we'll ignore not-found errors, since there is nothing to do.
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
-	log := ctrl.LoggerFrom(ctx).WithValues("clusterQueue", klog.KObj(&cqObj))
-	ctx = ctrl.LoggerInto(ctx, log)
-	log.V(2).Info("Reconciling ClusterQueue")
+
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile ClusterQueue")
 
 	if cqObj.ObjectMeta.DeletionTimestamp.IsZero() {
 		// Although we'll add the finalizer via webhook mutation now, this is still useful
diff --git a/pkg/controller/core/cohort_controller.go b/pkg/controller/core/cohort_controller.go
index 534e53d33..54e398680 100644
--- a/pkg/controller/core/cohort_controller.go
+++ b/pkg/controller/core/cohort_controller.go
@@ -91,8 +91,9 @@ func (r *CohortReconciler) Generic(e event.GenericEvent) bool {
 //+kubebuilder:rbac:groups=kueue.x-k8s.io,resources=cohorts,verbs=get;list;watch;create;update;patch;delete
 
 func (r *CohortReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	log := ctrl.LoggerFrom(ctx).WithValues("cohort", req.Name)
-	log.V(2).Info("Reconciling Cohort")
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile Cohort")
+
 	var cohort kueue.Cohort
 	if err := r.client.Get(ctx, req.NamespacedName, &cohort); err != nil {
 		if apierrors.IsNotFound(err) {
diff --git a/pkg/controller/core/localqueue_controller.go b/pkg/controller/core/localqueue_controller.go
index cc0fe407a..2652136d6 100644
--- a/pkg/controller/core/localqueue_controller.go
+++ b/pkg/controller/core/localqueue_controller.go
@@ -104,9 +104,9 @@ func (r *LocalQueueReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		// we'll ignore not-found errors, since there is nothing to do.
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
-	log := ctrl.LoggerFrom(ctx).WithValues("localQueue", klog.KObj(&queueObj))
-	ctx = ctrl.LoggerInto(ctx, log)
-	log.V(2).Info("Reconciling LocalQueue")
+
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile LocalQueue")
 
 	if ptr.Deref(queueObj.Spec.StopPolicy, kueue.None) != kueue.None {
 		err := r.UpdateStatusIfChanged(ctx, &queueObj, metav1.ConditionFalse, StoppedReason, localQueueIsInactiveMsg)
diff --git a/pkg/controller/core/resourceflavor_controller.go b/pkg/controller/core/resourceflavor_controller.go
index 812a0db1d..7addc739c 100644
--- a/pkg/controller/core/resourceflavor_controller.go
+++ b/pkg/controller/core/resourceflavor_controller.go
@@ -76,9 +76,9 @@ func (r *ResourceFlavorReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		// we'll ignore not-found errors, since there is nothing to do.
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
-	log := ctrl.LoggerFrom(ctx).WithValues("resourceFlavor", klog.KObj(&flavor))
-	ctx = ctrl.LoggerInto(ctx, log)
-	log.V(2).Info("Reconciling ResourceFlavor")
+
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile ResourceFlavor")
 
 	if flavor.DeletionTimestamp.IsZero() {
 		// Although we'll add the finalizer via webhook mutation now, this is still useful
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index b44eab420..f5e160924 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -140,9 +140,9 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		// we'll ignore not-found errors, since there is nothing to do.
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
-	log := ctrl.LoggerFrom(ctx).WithValues("workload", klog.KObj(&wl))
-	ctx = ctrl.LoggerInto(ctx, log)
-	log.V(2).Info("Reconciling Workload")
+
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile Workload")
 
 	if len(wl.ObjectMeta.OwnerReferences) == 0 && !wl.DeletionTimestamp.IsZero() {
 		return ctrl.Result{}, workload.RemoveFinalizer(ctx, r.client, &wl)
@@ -826,9 +826,17 @@ func (r *WorkloadReconciler) admittedNotReadyWorkload(wl *kueue.Workload) (bool,
 	if podsReadyCond != nil && podsReadyCond.Status == metav1.ConditionTrue {
 		return false, 0
 	}
-	admittedCond := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadAdmitted)
-	elapsedTime := r.clock.Since(admittedCond.LastTransitionTime.Time)
-	return true, max(r.waitForPodsReady.timeout-elapsedTime, 0)
+
+	if podsReadyCond == nil || podsReadyCond.Reason == kueue.WorkloadWaitForStart || podsReadyCond.Reason == "PodsReady" {
+		admittedCond := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadAdmitted)
+		elapsedTime := r.clock.Since(admittedCond.LastTransitionTime.Time)
+		return true, max(r.waitForPodsReady.timeout-elapsedTime, 0)
+	} else if podsReadyCond.Reason == kueue.WorkloadWaitForRecovery && r.waitForPodsReady.recoveryTimeout != nil {
+		// A pod has failed and the workload is waiting for recovery
+		elapsedTime := r.clock.Since(podsReadyCond.LastTransitionTime.Time)
+		return true, max(*r.waitForPodsReady.recoveryTimeout-elapsedTime, 0)
+	}
+	return false, 0
 }
 
 type resourceUpdatesHandler struct {
diff --git a/pkg/controller/jobframework/defaults.go b/pkg/controller/jobframework/defaults.go
index c7510ad4e..e4c20ac03 100644
--- a/pkg/controller/jobframework/defaults.go
+++ b/pkg/controller/jobframework/defaults.go
@@ -22,7 +22,6 @@ import (
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -52,7 +51,7 @@ func ApplyDefaultForSuspend(ctx context.Context, job GenericJob, k8sClient clien
 func WorkloadShouldBeSuspended(ctx context.Context, jobObj client.Object, k8sClient client.Client,
 	manageJobsWithoutQueueName bool, managedJobsNamespaceSelector labels.Selector) (bool, error) {
 	// Do not default suspend a job whose owner is already managed by Kueue
-	if owner := metav1.GetControllerOf(jobObj); owner != nil && IsOwnerManagedByKueue(owner) {
+	if IsOwnerManagedByKueueForObject(jobObj) {
 		return false, nil
 	}
 
@@ -85,7 +84,7 @@ func ApplyDefaultLocalQueue(jobObj client.Object, defaultQueueExist func(string)
 	}
 	if QueueNameForObject(jobObj) == "" {
 		// Do not default the queue-name for a job whose owner is already managed by Kueue
-		if owner := metav1.GetControllerOf(jobObj); owner != nil && IsOwnerManagedByKueue(owner) {
+		if IsOwnerManagedByKueueForObject(jobObj) {
 			return
 		}
 		labels := jobObj.GetLabels()
diff --git a/pkg/controller/jobframework/integrationmanager.go b/pkg/controller/jobframework/integrationmanager.go
index 111329a78..ad9d60459 100644
--- a/pkg/controller/jobframework/integrationmanager.go
+++ b/pkg/controller/jobframework/integrationmanager.go
@@ -318,6 +318,15 @@ func IsOwnerManagedByKueue(owner *metav1.OwnerReference) bool {
 	return manager.getJobTypeForOwner(owner) != nil
 }
 
+// IsOwnerManagedByKueueForObject returns true if the provided object has an owner,
+// and this owner can be managed by Kueue.
+func IsOwnerManagedByKueueForObject(obj client.Object) bool {
+	if owner := metav1.GetControllerOf(obj); owner != nil {
+		return IsOwnerManagedByKueue(owner)
+	}
+	return false
+}
+
 // IsOwnerIntegrationEnabled returns true if the provided owner is managed by an enabled integration.
 func IsOwnerIntegrationEnabled(owner *metav1.OwnerReference) bool {
 	// This function should be redundant with IsOwnerManagedByKueue, but currently is not.
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index 62e16d00d..e1cbe8854 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -460,15 +460,17 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 	// handle a job when waitForPodsReady is enabled, and it is the main job
 	if r.waitForPodsReady {
 		log.V(3).Info("Handling a job when waitForPodsReady is enabled")
-		condition := generatePodsReadyCondition(job, wl)
-		// optimization to avoid sending the update request if the status didn't change
-		if !apimeta.IsStatusConditionPresentAndEqual(wl.Status.Conditions, condition.Type, condition.Status) {
-			log.V(3).Info(fmt.Sprintf("Updating the PodsReady condition with status: %v", condition.Status))
+		condition := generatePodsReadyCondition(log, job, wl)
+		if !workload.HasConditionWithTypeAndReason(wl, &condition) {
+			log.V(3).Info("Updating the PodsReady condition", "reason", condition.Reason, "status", condition.Status)
 			apimeta.SetStatusCondition(&wl.Status.Conditions, condition)
 			err := workload.UpdateStatus(ctx, r.client, wl, condition.Type, condition.Status, condition.Reason, condition.Message, constants.JobControllerName, r.clock)
 			if err != nil {
 				log.Error(err, "Updating workload status")
 			}
+			return ctrl.Result{}, nil
+		} else {
+			log.V(3).Info("No update for PodsReady condition")
 		}
 	}
 
@@ -1148,24 +1150,57 @@ func (r *JobReconciler) ignoreUnretryableError(log logr.Logger, err error) error
 	return err
 }
 
-func generatePodsReadyCondition(job GenericJob, wl *kueue.Workload) metav1.Condition {
-	conditionStatus := metav1.ConditionFalse
-	message := "Not all pods are ready or succeeded"
-	// Once PodsReady=True it stays as long as the workload remains admitted to
-	// avoid unnecessary flickering the condition when the pods transition
-	// from Ready to Completed. As pods finish, they transition first into the
-	// uncountedTerminatedPods staging area, before passing to the
-	// succeeded/failed counters.
-	if workload.IsAdmitted(wl) && (job.PodsReady() || apimeta.IsStatusConditionTrue(wl.Status.Conditions, kueue.WorkloadPodsReady)) {
-		conditionStatus = metav1.ConditionTrue
-		message = "All pods were ready or succeeded since the workload admission"
-	}
-	return metav1.Condition{
-		Type:    kueue.WorkloadPodsReady,
-		Status:  conditionStatus,
-		Reason:  "PodsReady",
-		Message: message,
-		// ObservedGeneration is added via workload.UpdateStatus
+func generatePodsReadyCondition(log logr.Logger, job GenericJob, wl *kueue.Workload) metav1.Condition {
+	const (
+		notReadyMsg           = "Not all pods are ready or succeeded"
+		waitingForRecoveryMsg = "At least one pod has failed, waiting for recovery"
+		readyMsg              = "All pods reached readiness and the workload is running"
+	)
+	if !workload.IsAdmitted(wl) {
+		// The workload has not been admitted yet
+		// or it was admitted in the past but it's evicted/requeued
+		return workload.CreatePodsReadyCondition(metav1.ConditionFalse,
+			kueue.WorkloadWaitForStart,
+			notReadyMsg)
+	}
+
+	podsReadyCond := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadPodsReady)
+	podsReady := job.PodsReady()
+	log.V(3).Info("Generating PodsReady condition",
+		"Current PodsReady condition", podsReadyCond,
+		"Pods are ready", podsReady)
+
+	if podsReady {
+		reason := kueue.WorkloadStarted
+		if podsReadyCond != nil && (podsReadyCond.Reason == kueue.WorkloadWaitForRecovery || podsReadyCond.Reason == kueue.WorkloadRecovered) {
+			reason = kueue.WorkloadRecovered
+		}
+		return workload.CreatePodsReadyCondition(metav1.ConditionTrue,
+			reason,
+			readyMsg)
+	}
+
+	switch {
+	case podsReadyCond == nil:
+		return workload.CreatePodsReadyCondition(metav1.ConditionFalse,
+			kueue.WorkloadWaitForStart,
+			notReadyMsg)
+
+	case podsReadyCond.Status == metav1.ConditionTrue:
+		return workload.CreatePodsReadyCondition(metav1.ConditionFalse,
+			kueue.WorkloadWaitForRecovery,
+			waitingForRecoveryMsg)
+
+	case podsReadyCond.Reason == kueue.WorkloadWaitForRecovery:
+		return workload.CreatePodsReadyCondition(metav1.ConditionFalse,
+			kueue.WorkloadWaitForRecovery,
+			waitingForRecoveryMsg)
+
+	default:
+		// handles both "WaitForPodsStart" and the old "PodsReady" reasons
+		return workload.CreatePodsReadyCondition(metav1.ConditionFalse,
+			kueue.WorkloadWaitForStart,
+			notReadyMsg)
 	}
 }
 
diff --git a/pkg/controller/jobframework/validation.go b/pkg/controller/jobframework/validation.go
index 3b4f7e148..aee71b0c4 100644
--- a/pkg/controller/jobframework/validation.go
+++ b/pkg/controller/jobframework/validation.go
@@ -30,7 +30,6 @@ import (
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	apivalidation "k8s.io/apimachinery/pkg/api/validation"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
@@ -207,10 +206,10 @@ func validateImmutablePodGroupPodSpecPath(newShape, oldShape map[string]interfac
 }
 
 func IsManagedByKueue(obj client.Object) bool {
-	objectOwner := metav1.GetControllerOf(obj)
-	if objectOwner != nil && IsOwnerManagedByKueue(objectOwner) {
+	if IsOwnerManagedByKueueForObject(obj) {
 		return false
-	} else if QueueNameForObject(obj) != "" {
+	}
+	if QueueNameForObject(obj) != "" {
 		return true
 	}
 	return false
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
index 8f143862e..aa03af063 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_pod_reconciler.go
@@ -67,9 +67,8 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	log := ctrl.LoggerFrom(ctx).WithValues("pod", klog.KObj(pod))
-	ctx = ctrl.LoggerInto(ctx, log)
-	log.V(2).Info("Reconciling LeaderWorkerSet Pod")
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile LeaderWorkerSet Pod")
 
 	if utilpod.IsTerminated(pod) {
 		err = client.IgnoreNotFound(clientutil.Patch(ctx, r.client, pod, true, func() (bool, error) {
diff --git a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
index 9ea18f550..9bf1e0d47 100644
--- a/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
+++ b/pkg/controller/jobs/leaderworkerset/leaderworkerset_reconciler.go
@@ -22,7 +22,6 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/record"
-	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -80,9 +79,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	log := ctrl.LoggerFrom(ctx).WithValues("leaderworkerset", klog.KObj(lws))
-	ctx = ctrl.LoggerInto(ctx, log)
-	log.V(2).Info("Reconciling LeaderWorkerSet")
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile LeaderWorkerSet")
 
 	err = r.createPrebuiltWorkloadsIfNotExist(ctx, lws)
 	if err != nil {
diff --git a/pkg/controller/jobs/pod/pod_controller.go b/pkg/controller/jobs/pod/pod_controller.go
index 673c32865..a26e893b0 100644
--- a/pkg/controller/jobs/pod/pod_controller.go
+++ b/pkg/controller/jobs/pod/pod_controller.go
@@ -1329,13 +1329,6 @@ func (p *Pod) EquivalentToWorkload(ctx context.Context, c client.Client, wl *kue
 	return p.equivalentToWorkload(wl, podSets), nil
 }
 
-func IsPodOwnerManagedByKueue(p *Pod) bool {
-	if owner := metav1.GetControllerOf(&p.pod); owner != nil {
-		return jobframework.IsOwnerManagedByKueue(owner)
-	}
-	return false
-}
-
 func GetWorkloadNameForPod(podName string, podUID types.UID) string {
 	return jobframework.GetWorkloadNameForOwnerWithGVK(podName, podUID, gvk)
 }
diff --git a/pkg/controller/jobs/pod/pod_webhook.go b/pkg/controller/jobs/pod/pod_webhook.go
index 272ad563a..15099d540 100644
--- a/pkg/controller/jobs/pod/pod_webhook.go
+++ b/pkg/controller/jobs/pod/pod_webhook.go
@@ -308,7 +308,8 @@ func validateManagedLabel(pod *Pod) field.ErrorList {
 
 // warningForPodManagedLabel returns a warning message if the pod has a managed label, and it's parent is managed by kueue
 func warningForPodManagedLabel(p *Pod) string {
-	if managedLabel := p.pod.GetLabels()[ManagedLabelKey]; managedLabel == ManagedLabelValue && IsPodOwnerManagedByKueue(p) {
+	managedLabel := p.pod.GetLabels()[ManagedLabelKey]
+	if managedLabel == ManagedLabelValue && jobframework.IsOwnerManagedByKueueForObject(p.Object()) {
 		return fmt.Sprintf("pod owner is managed by kueue, label '%s=%s' might lead to unexpected behaviour",
 			ManagedLabelKey, ManagedLabelValue)
 	}
diff --git a/pkg/controller/jobs/statefulset/statefulset_reconciler.go b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
index 6acc34950..bee918085 100644
--- a/pkg/controller/jobs/statefulset/statefulset_reconciler.go
+++ b/pkg/controller/jobs/statefulset/statefulset_reconciler.go
@@ -57,9 +57,8 @@ type Reconciler struct {
 }
 
 func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
-	log := ctrl.LoggerFrom(ctx).WithValues("statefulset", klog.KRef(req.Namespace, req.Name))
-	ctx = ctrl.LoggerInto(ctx, log)
-	log.V(2).Info("Reconciling StatefulSet")
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Reconcile StatefulSet")
 
 	err := r.fetchAndFinalizePods(ctx, req)
 	return ctrl.Result{}, err
diff --git a/pkg/controller/tas/resource_flavor.go b/pkg/controller/tas/resource_flavor.go
index 55139bac5..0e653a439 100644
--- a/pkg/controller/tas/resource_flavor.go
+++ b/pkg/controller/tas/resource_flavor.go
@@ -134,7 +134,7 @@ func (h *nodeHandler) Generic(context.Context, event.GenericEvent, workqueue.Typ
 }
 
 func (r *rfReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
-	log := ctrl.LoggerFrom(ctx).WithValues("name", req.NamespacedName.Name)
+	log := ctrl.LoggerFrom(ctx)
 	log.V(2).Info("Reconcile TAS Resource Flavor")
 
 	flv := &kueue.ResourceFlavor{}
diff --git a/pkg/controller/tas/topology_ungater.go b/pkg/controller/tas/topology_ungater.go
index c7dabcbd4..dc181a6ee 100644
--- a/pkg/controller/tas/topology_ungater.go
+++ b/pkg/controller/tas/topology_ungater.go
@@ -147,7 +147,7 @@ func (h *podHandler) queueReconcileForPod(ctx context.Context, object client.Obj
 }
 
 func (r *topologyUngater) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
-	log := ctrl.LoggerFrom(ctx).WithValues("workload", req.NamespacedName.String())
+	log := ctrl.LoggerFrom(ctx)
 	log.V(2).Info("Reconcile Topology Ungater")
 
 	wl := &kueue.Workload{}
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 172e13cd8..8dac52e1b 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -224,11 +224,11 @@ func (p *Preemptor) IssuePreemptions(ctx context.Context, preemptor *workload.In
 				return
 			}
 
-			log.V(3).Info("Preempted", "targetWorkload", klog.KObj(target.WorkloadInfo.Obj), "reason", target.Reason, "message", message, "targetClusterQueue", klog.KRef("", target.WorkloadInfo.ClusterQueue))
+			log.V(3).Info("Preempted", "targetWorkload", klog.KObj(target.WorkloadInfo.Obj), "preemptingWorkload", klog.KObj(preemptor.Obj), "reason", target.Reason, "message", message, "targetClusterQueue", klog.KRef("", target.WorkloadInfo.ClusterQueue))
 			p.recorder.Eventf(target.WorkloadInfo.Obj, corev1.EventTypeNormal, "Preempted", message)
 			metrics.ReportPreemption(preemptor.ClusterQueue, target.Reason, target.WorkloadInfo.ClusterQueue)
 		} else {
-			log.V(3).Info("Preemption ongoing", "targetWorkload", klog.KObj(target.WorkloadInfo.Obj))
+			log.V(3).Info("Preemption ongoing", "targetWorkload", klog.KObj(target.WorkloadInfo.Obj), "preemptingWorkload", klog.KObj(preemptor.Obj))
 		}
 		successfullyPreempted.Add(1)
 	})
@@ -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)
+		logV.Info("Simulating preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption, "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)
+		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption, "allowBorrowingBelowPriority", allowBorrowingBelowPriority, "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
 	}
 	requests := preemptionCtx.requests
 	cqHeap := cqHeapFromCandidates(candidates, false, preemptionCtx.snapshot)
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 78225a967..f7425ef92 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -798,6 +798,28 @@ func IsEvicted(w *kueue.Workload) bool {
 	return apimeta.IsStatusConditionPresentAndEqual(w.Status.Conditions, kueue.WorkloadEvicted, metav1.ConditionTrue)
 }
 
+// HasConditionWithTypeAndReason checks if there is a condition in Workload's status
+// with exactly the same Type, Status and Reason
+func HasConditionWithTypeAndReason(w *kueue.Workload, cond *metav1.Condition) bool {
+	for _, statusCond := range w.Status.Conditions {
+		if statusCond.Type == cond.Type && statusCond.Reason == cond.Reason &&
+			statusCond.Status == cond.Status {
+			return true
+		}
+	}
+	return false
+}
+
+func CreatePodsReadyCondition(status metav1.ConditionStatus, reason, message string) metav1.Condition {
+	return metav1.Condition{
+		Type:    kueue.WorkloadPodsReady,
+		Status:  status,
+		Reason:  reason,
+		Message: message,
+		// ObservedGeneration is added via workload.UpdateStatus
+	}
+}
+
 func RemoveFinalizer(ctx context.Context, c client.Client, wl *kueue.Workload) error {
 	if controllerutil.RemoveFinalizer(wl, kueue.ResourceInUseFinalizerName) {
 		return c.Update(ctx, wl)
diff --git a/site/content/en/docs/concepts/topology_aware_scheduling.md b/site/content/en/docs/concepts/topology_aware_scheduling.md
index 4067cfff1..10c044097 100644
--- a/site/content/en/docs/concepts/topology_aware_scheduling.md
+++ b/site/content/en/docs/concepts/topology_aware_scheduling.md
@@ -18,7 +18,7 @@ We observe that the data centers have a hierarchical structure of their
 organizational units, like racks and blocks, where there are multiple nodes
 within a rack, and there are multiple racks within a block. Pods running within
 the same organizational unit have better network bandwidth than Pods on
-different units. We say that nods placed in different racks are more distant
+different units. We say that nodes placed in different racks are more distant
 than nodes placed within the same rack. Similarly, nodes placed in different
 blocks are more distant than two nodes within the same block.
 
diff --git a/site/content/en/docs/reference/kueue.v1beta1.md b/site/content/en/docs/reference/kueue.v1beta1.md
index c1b8c50bf..0878cc902 100644
--- a/site/content/en/docs/reference/kueue.v1beta1.md
+++ b/site/content/en/docs/reference/kueue.v1beta1.md
@@ -874,7 +874,7 @@ reclaim its nominal quota.</li>
 and there are admitted Workloads in the ClusterQueue with lower priority.</li>
 </ul>
 <p>The preemption algorithm tries to find a minimal set of Workloads to
-preempt to accomomdate the pending Workload, preempting Workloads with
+preempt to accommodate the pending Workload, preempting Workloads with
 lower priority first.</p>
 </td>
 </tr>
@@ -2133,7 +2133,9 @@ controller that integrates with the Workload object.</p>
    <p>nodeTaints are taints that the nodes associated with this ResourceFlavor
 have.
 Workloads' podsets must have tolerations for these nodeTaints in order to
-get assigned this ResourceFlavor during admission.</p>
+get assigned this ResourceFlavor during admission.
+Only the 'NoSchedule' and 'NoExecute' taint effects are evaluated,
+while 'PreferNoSchedule' is ignored.</p>
 <p>An example of a nodeTaint is
 cloud.provider.com/preemptible=&quot;true&quot;:NoSchedule</p>
 <p>nodeTaints can be up to 8 elements.</p>
diff --git a/site/hugo.toml b/site/hugo.toml
index 6b4bfe691..ab5f4efba 100644
--- a/site/hugo.toml
+++ b/site/hugo.toml
@@ -90,7 +90,7 @@ ignoreFiles = []
   # The major.minor version tag for the version of the docs represented in this
   # branch of the repository. Used in the "version-banner" partial to display a
   # version number for this doc set.
-  version = "v0.10.1"
+  version = "v0.10.2"
 
   # Flag used in the "version-banner" partial to decide whether to display a
   # banner on every page indicating that this is an archived version of the docs.



Test patch:
diff --git a/pkg/controller/core/workload_controller_test.go b/pkg/controller/core/workload_controller_test.go
index 621ff8cc4..227bf0b60 100644
--- a/pkg/controller/core/workload_controller_test.go
+++ b/pkg/controller/core/workload_controller_test.go
@@ -102,8 +102,9 @@ func TestAdmittedNotReadyWorkload(t *testing.T) {
 			},
 			waitForPodsReady:           &waitForPodsReadyConfig{timeout: 5 * time.Minute},
 			wantCountingTowardsTimeout: true,
+			wantRecheckAfter:           0,
 		},
-		"workload with Admitted=True, PodsReady=False; counting since PodsReady.LastTransitionTime": {
+		"with reason WorkloadWaitForPodsReadyStart; workload with Admitted=True, PodsReady=False; counting since admitted.LastTransitionTime": {
 			workload: kueue.Workload{
 				Status: kueue.WorkloadStatus{
 					Admission: &kueue.Admission{},
@@ -116,6 +117,7 @@ func TestAdmittedNotReadyWorkload(t *testing.T) {
 						{
 							Type:               kueue.WorkloadPodsReady,
 							Status:             metav1.ConditionFalse,
+							Reason:             kueue.WorkloadWaitForStart,
 							LastTransitionTime: metav1.NewTime(now),
 						},
 					},
@@ -125,6 +127,74 @@ func TestAdmittedNotReadyWorkload(t *testing.T) {
 			wantCountingTowardsTimeout: true,
 			wantRecheckAfter:           4 * time.Minute,
 		},
+		"with reason PodsReady; workload with Admitted=True, PodsReady=False; counting since admitted.LastTransitionTime": {
+			workload: kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					Admission: &kueue.Admission{},
+					Conditions: []metav1.Condition{
+						{
+							Type:               kueue.WorkloadAdmitted,
+							Status:             metav1.ConditionTrue,
+							LastTransitionTime: metav1.NewTime(minuteAgo),
+						},
+						{
+							Type:               kueue.WorkloadPodsReady,
+							Status:             metav1.ConditionFalse,
+							Reason:             "PodsReady",
+							LastTransitionTime: metav1.NewTime(now),
+						},
+					},
+				},
+			},
+			waitForPodsReady:           &waitForPodsReadyConfig{timeout: 5 * time.Minute},
+			wantCountingTowardsTimeout: true,
+			wantRecheckAfter:           4 * time.Minute,
+		},
+		"workload with Admitted=True, PodsReady=False, Reason=WorkloadWaitForPodsReadyRecovery": {
+			workload: kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					Admission: &kueue.Admission{},
+					Conditions: []metav1.Condition{
+						{
+							Type:               kueue.WorkloadAdmitted,
+							Status:             metav1.ConditionTrue,
+							LastTransitionTime: metav1.NewTime(minuteAgo),
+						},
+						{
+							Type:               kueue.WorkloadPodsReady,
+							Status:             metav1.ConditionFalse,
+							Reason:             kueue.WorkloadWaitForRecovery,
+							LastTransitionTime: metav1.NewTime(now),
+						},
+					},
+				},
+			},
+			waitForPodsReady:           &waitForPodsReadyConfig{recoveryTimeout: ptr.To(3 * time.Minute)},
+			wantCountingTowardsTimeout: true,
+			wantRecheckAfter:           3 * time.Minute,
+		},
+		"workload with Admitted=True, PodsReady=False, Reason=WorkloadWaitForPodsReadyRecovery, recoveryTimeout not configured": {
+			workload: kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					Admission: &kueue.Admission{},
+					Conditions: []metav1.Condition{
+						{
+							Type:               kueue.WorkloadAdmitted,
+							Status:             metav1.ConditionTrue,
+							LastTransitionTime: metav1.NewTime(minuteAgo),
+						},
+						{
+							Type:               kueue.WorkloadPodsReady,
+							Status:             metav1.ConditionFalse,
+							Reason:             kueue.WorkloadWaitForRecovery,
+							LastTransitionTime: metav1.NewTime(now),
+						},
+					},
+				},
+			},
+			waitForPodsReady:           &waitForPodsReadyConfig{recoveryTimeout: nil},
+			wantCountingTowardsTimeout: false,
+		},
 		"workload with Admitted=Unknown; not counting": {
 			workload: kueue.Workload{
 				Status: kueue.WorkloadStatus{
@@ -1064,7 +1134,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 		},
-		"should set the Evicted condition with Deactivated reason, exceeded the maximum number of requeue retries" +
+		"with reason PodsReady; should set the Evicted condition with Deactivated reason, exceeded the maximum number of requeue retries" +
 			"when the .spec.active is False, Admitted, the Workload has Evicted=False and DeactivationTarget=True condition": {
 			reconcilerOpts: []Option{
 				WithWaitForPodsReady(&waitForPodsReadyConfig{
@@ -1124,7 +1194,67 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 		},
-		"[backoffLimitCount: 100] should set the Evicted condition with Deactivated reason, exceeded the maximum number of requeue retries" +
+		"with reason WaitForPodsStart; should set the Evicted condition with Deactivated reason, exceeded the maximum number of requeue retries" +
+			"when the .spec.active is False, Admitted, the Workload has Evicted=False and DeactivationTarget=True condition": {
+			reconcilerOpts: []Option{
+				WithWaitForPodsReady(&waitForPodsReadyConfig{
+					timeout:                     3 * time.Second,
+					requeuingBackoffLimitCount:  ptr.To[int32](0),
+					requeuingBackoffBaseSeconds: 10,
+					requeuingBackoffJitter:      0,
+				}),
+			},
+			workload: utiltesting.MakeWorkload("wl", "ns").
+				Active(false).
+				ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadDeactivationTarget,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadRequeuingLimitExceeded,
+					Message: "exceeding the maximum number of re-queuing retries",
+				}).
+				Obj(),
+			wantWorkload: utiltesting.MakeWorkload("wl", "ns").
+				Active(false).
+				ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadEvicted,
+					Status:  metav1.ConditionTrue,
+					Reason:  "DeactivatedDueToRequeuingLimitExceeded",
+					Message: "The workload is deactivated due to exceeding the maximum number of re-queuing retries",
+				}).
+				// DeactivationTarget condition should be deleted in the real cluster, but the fake client doesn't allow us to do it.
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadDeactivationTarget,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadRequeuingLimitExceeded,
+					Message: "exceeding the maximum number of re-queuing retries",
+				}).
+				Obj(),
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Name: "wl", Namespace: "ns"},
+					EventType: corev1.EventTypeNormal,
+					Reason:    "EvictedDueToDeactivatedDueToRequeuingLimitExceeded",
+					Message:   "The workload is deactivated due to exceeding the maximum number of re-queuing retries",
+				},
+			},
+		},
+		"with reason PodsReady; [backoffLimitCount: 100] should set the Evicted condition with Deactivated reason, exceeded the maximum number of requeue retries" +
 			"when the .spec.active is False, Admitted, the Workload has Evicted=False and DeactivationTarget=True condition, and the requeueState.count equals to backoffLimitCount": {
 			reconcilerOpts: []Option{
 				WithWaitForPodsReady(&waitForPodsReadyConfig{
@@ -1187,6 +1317,69 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 		},
+		"with reason WaitForPodsStart; [backoffLimitCount: 100] should set the Evicted condition with Deactivated reason, exceeded the maximum number of requeue retries" +
+			"when the .spec.active is False, Admitted, the Workload has Evicted=False and DeactivationTarget=True condition, and the requeueState.count equals to backoffLimitCount": {
+			reconcilerOpts: []Option{
+				WithWaitForPodsReady(&waitForPodsReadyConfig{
+					timeout:                     3 * time.Second,
+					requeuingBackoffLimitCount:  ptr.To[int32](100),
+					requeuingBackoffBaseSeconds: 10,
+					requeuingBackoffJitter:      0,
+				}),
+			},
+			workload: utiltesting.MakeWorkload("wl", "ns").
+				Active(false).
+				ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadDeactivationTarget,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadRequeuingLimitExceeded,
+					Message: "exceeding the maximum number of re-queuing retries",
+				}).
+				RequeueState(ptr.To[int32](100), nil).
+				Obj(),
+			wantWorkload: utiltesting.MakeWorkload("wl", "ns").
+				Active(false).
+				ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadEvicted,
+					Status:  metav1.ConditionTrue,
+					Reason:  "DeactivatedDueToRequeuingLimitExceeded",
+					Message: "The workload is deactivated due to exceeding the maximum number of re-queuing retries",
+				}).
+				// DeactivationTarget condition should be deleted in the real cluster, but the fake client doesn't allow us to do it.
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadDeactivationTarget,
+					Status:  metav1.ConditionTrue,
+					Reason:  "RequeuingLimitExceeded",
+					Message: "exceeding the maximum number of re-queuing retries",
+				}).
+				// The requeueState should be reset in the real cluster, but the fake client doesn't allow us to do it.
+				RequeueState(ptr.To[int32](100), nil).
+				Obj(),
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Name: "wl", Namespace: "ns"},
+					EventType: corev1.EventTypeNormal,
+					Reason:    "EvictedDueToDeactivatedDueToRequeuingLimitExceeded",
+					Message:   "The workload is deactivated due to exceeding the maximum number of re-queuing retries",
+				},
+			},
+		},
 		"should keep the previous eviction reason when the Workload is already evicted by other reason even though the Workload is deactivated.": {
 			workload: utiltesting.MakeWorkload("wl", "ns").
 				Active(false).
diff --git a/pkg/controller/jobs/job/job_controller_test.go b/pkg/controller/jobs/job/job_controller_test.go
index 126fc3c43..eb84d0a30 100644
--- a/pkg/controller/jobs/job/job_controller_test.go
+++ b/pkg/controller/jobs/job/job_controller_test.go
@@ -35,6 +35,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
@@ -470,6 +471,8 @@ func TestReconciler(t *testing.T) {
 		},
 	}
 
+	baseWaitForPodsReadyConf := &configapi.WaitForPodsReady{Enable: true}
+
 	cases := map[string]struct {
 		enableTopologyAwareScheduling bool
 
@@ -483,6 +486,323 @@ func TestReconciler(t *testing.T) {
 		wantEvents        []utiltesting.EventRecord
 		wantErr           error
 	}{
+		"PodsReady is set to False before Workload is Admitted": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job:     *baseJobWrapper.DeepCopy(),
+			wantJob: *baseJobWrapper.DeepCopy(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(false).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(false).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady is set to False after Workload is Admitted but not all Pods reached readiness": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job:     *baseJobWrapper.DeepCopy(),
+			wantJob: *baseJobWrapper.DeepCopy(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady is set to False after Workload is Admitted, some Pods became ready but not all Pods reached readiness": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job: *baseJobWrapper.Clone().
+				Suspend(false).
+				Ready(9).
+				Active(10).
+				Obj(),
+			wantJob: *baseJobWrapper.Clone().
+				Suspend(false).
+				Ready(9).
+				Active(10).
+				Obj(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady is set to True after Workload is Admitted and all Pods reached readiness": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			wantJob: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadStarted,
+					Message: "All pods reached readiness and the workload is running",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady is set to True after Workload is Admitted and all Pods reached readiness without previous PodsReady condition": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			wantJob: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadStarted,
+					Message: "All pods reached readiness and the workload is running",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady is set to False after Workload is running and one pod failed": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job: *baseJobWrapper.Clone().
+				Ready(9).
+				Failed(1).
+				Obj(),
+			wantJob: *baseJobWrapper.Clone().
+				Ready(9).
+				Failed(1).
+				Obj(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadStarted,
+					Message: "All pods reached readiness and the workload is running",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForRecovery,
+					Message: "At least one pod has failed, waiting for recovery",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady continues to be False after a pod failed and workload is still recovering": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job: *baseJobWrapper.Clone().
+				Suspend(false).
+				Ready(9).
+				Failed(1).
+				Obj(),
+			wantJob: *baseJobWrapper.Clone().
+				Suspend(false).
+				Ready(9).
+				Failed(1).
+				Obj(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForRecovery,
+					Message: "At least one pod has failed, waiting for recovery",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForRecovery,
+					Message: "At least one pod has failed, waiting for recovery",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady is set to True after failing pod recovered": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			wantJob: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForRecovery,
+					Message: "At least one pod has failed, waiting for recovery",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadRecovered,
+					Message: "All pods reached readiness and the workload is running",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady=False has the new Reason if there was the old one before (pre v0.11.0)": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job:     *baseJobWrapper.DeepCopy(),
+			wantJob: *baseJobWrapper.DeepCopy(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  "PodsReady",
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady=True has the new Reason if there was the old one before (pre v0.11.0)": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			wantJob: *baseJobWrapper.Clone().
+				Ready(10).
+				Obj(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionTrue,
+					Reason:  "PodsReady",
+					Message: "All pods reached readiness and the workload is running",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionTrue,
+					Reason:  kueue.WorkloadStarted,
+					Message: "All pods reached readiness and the workload is running",
+				}).
+				Obj(),
+			},
+		},
+		"PodsReady is set to False if there's an invalid Reason (pre v0.11.0)": {
+			reconcilerOptions: []jobframework.Option{
+				jobframework.WithWaitForPodsReady(baseWaitForPodsReadyConf),
+			},
+			job:     *baseJobWrapper.DeepCopy(),
+			wantJob: *baseJobWrapper.DeepCopy(),
+			workloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  "InvalidReason",
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+			wantWorkloads: []kueue.Workload{*baseWorkloadWrapper.Clone().
+				Admitted(true).
+				Condition(metav1.Condition{
+					Type:    kueue.WorkloadPodsReady,
+					Status:  metav1.ConditionFalse,
+					Reason:  kueue.WorkloadWaitForStart,
+					Message: "Not all pods are ready or succeeded",
+				}).
+				Obj(),
+			},
+		},
 		"PodSet label and Workload annotation are set when Job is starting; TopologyAwareScheduling enabled": {
 			enableTopologyAwareScheduling: true,
 			reconcilerOptions: []jobframework.Option{
diff --git a/pkg/controller/jobs/pod/pod_controller_test.go b/pkg/controller/jobs/pod/pod_controller_test.go
index 6f3f6cdc2..62ad60945 100644
--- a/pkg/controller/jobs/pod/pod_controller_test.go
+++ b/pkg/controller/jobs/pod/pod_controller_test.go
@@ -5642,9 +5642,8 @@ func TestIsPodOwnerManagedByQueue(t *testing.T) {
 
 			pod.OwnerReferences = append(pod.OwnerReferences, tc.ownerReference)
 
-			if tc.wantRes != IsPodOwnerManagedByKueue(FromObject(pod)) {
-				t.Errorf("Unexpected 'IsPodOwnerManagedByKueue' result\n want: %t\n got: %t)",
-					tc.wantRes, IsPodOwnerManagedByKueue(FromObject(pod)))
+			if managedByKueue := jobframework.IsOwnerManagedByKueueForObject(pod); tc.wantRes != managedByKueue {
+				t.Errorf("Unexpected 'IsOwnerManagedByKueueForObject' result\n want: %t\n got: %t)", tc.wantRes, managedByKueue)
 			}
 		})
 	}
diff --git a/pkg/util/testing/condition_matchers.go b/pkg/util/testing/condition_matchers.go
index 36757b8f0..f7dfe826f 100644
--- a/pkg/util/testing/condition_matchers.go
+++ b/pkg/util/testing/condition_matchers.go
@@ -26,24 +26,30 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func HaveConditionStatus(condition string, status metav1.ConditionStatus) types.GomegaMatcher {
+func HaveConditionStatus(conditionType string, status metav1.ConditionStatus) types.GomegaMatcher {
 	return &conditionMatcher{
-		condition: condition,
-		status:    status,
+		conditionType: conditionType,
+		status:        status,
 	}
 }
 
-func HaveConditionStatusTrue(condition string) types.GomegaMatcher {
-	return HaveConditionStatus(condition, metav1.ConditionTrue)
+func HaveCondition(conditionType string) types.GomegaMatcher {
+	return &conditionMatcher{
+		conditionType: conditionType,
+	}
+}
+
+func HaveConditionStatusTrue(conditionType string) types.GomegaMatcher {
+	return HaveConditionStatus(conditionType, metav1.ConditionTrue)
 }
 
-func HaveConditionStatusFalse(condition string) types.GomegaMatcher {
-	return HaveConditionStatus(condition, metav1.ConditionFalse)
+func HaveConditionStatusFalse(conditionType string) types.GomegaMatcher {
+	return HaveConditionStatus(conditionType, metav1.ConditionFalse)
 }
 
 type conditionMatcher struct {
-	condition string
-	status    metav1.ConditionStatus
+	conditionType string
+	status        metav1.ConditionStatus
 }
 
 func (matcher *conditionMatcher) Match(actual interface{}) (bool, error) {
@@ -52,7 +58,7 @@ func (matcher *conditionMatcher) Match(actual interface{}) (bool, error) {
 		return false, fmt.Errorf("Condition matcher expects a []metav1.Condition. Got:\n%s", format.Object(actual, 1))
 	}
 
-	found := apimeta.FindStatusCondition(conditions, matcher.condition)
+	found := apimeta.FindStatusCondition(conditions, matcher.conditionType)
 	if found == nil {
 		return false, nil
 	}
@@ -80,8 +86,8 @@ func (matcher *conditionMatcher) buildErrorMessage(actual interface{}, negated b
 	if negated {
 		b.WriteString("not ")
 	}
-	b.WriteString("to have condition ")
-	b.WriteString(matcher.condition)
+	b.WriteString("to have condition type ")
+	b.WriteString(matcher.conditionType)
 	if matcher.status != "" {
 		b.WriteString(" and status ")
 		b.WriteString(string(matcher.status))
diff --git a/pkg/util/testingjobs/job/wrappers.go b/pkg/util/testingjobs/job/wrappers.go
index 946c410a4..c4b08d9d4 100644
--- a/pkg/util/testingjobs/job/wrappers.go
+++ b/pkg/util/testingjobs/job/wrappers.go
@@ -229,6 +229,18 @@ func (j *JobWrapper) Active(c int32) *JobWrapper {
 	return j
 }
 
+// Failed sets the .status.failed
+func (j *JobWrapper) Failed(c int32) *JobWrapper {
+	j.Status.Failed = c
+	return j
+}
+
+// Ready sets the .status.ready
+func (j *JobWrapper) Ready(c int32) *JobWrapper {
+	j.Status.Ready = &c
+	return j
+}
+
 // Condition adds a condition
 func (j *JobWrapper) Condition(c batchv1.JobCondition) *JobWrapper {
 	j.Status.Conditions = append(j.Status.Conditions, c)
diff --git a/test/e2e/customconfigs/skipjobswithoutqueuename_test.go b/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
index 615fc8102..2d922d7bb 100644
--- a/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
+++ b/test/e2e/customconfigs/skipjobswithoutqueuename_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
+	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -34,6 +35,7 @@ import (
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	awtesting "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
+	testingdeploy "sigs.k8s.io/kueue/pkg/util/testingjobs/deployment"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingjobset "sigs.k8s.io/kueue/pkg/util/testingjobs/jobset"
 	testingpod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
@@ -261,8 +263,8 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 			ginkgo.By("creating a pod without a queue name", func() {
 				testPod = testingpod.MakePod("test-pod", "kube-system").
 					Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
-					Request("cpu", "1").
-					Request("memory", "2G").
+					Request(corev1.ResourceCPU, "1").
+					Request(corev1.ResourceMemory, "2Gi").
 					Obj()
 				gomega.Expect(k8sClient.Create(ctx, testPod)).Should(gomega.Succeed())
 			})
@@ -283,5 +285,68 @@ var _ = ginkgo.Describe("ManageJobsWithoutQueueName", ginkgo.Ordered, func() {
 				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
 			})
 		})
+
+		ginkgo.It("should suspend the pods of a deployment created in the test namespace", func() {
+			var testDeploy *appsv1.Deployment
+			ginkgo.By("creating a deployment without a queue name", func() {
+				testDeploy = testingdeploy.MakeDeployment("test-deploy", ns.Name).
+					Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+					Request(corev1.ResourceCPU, "1").
+					Request(corev1.ResourceMemory, "2Gi").
+					Replicas(2).
+					Obj()
+				gomega.Expect(k8sClient.Create(ctx, testDeploy)).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying that the pods of the deployment are gated", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					pods := &corev1.PodList{}
+					g.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Namespace),
+						client.MatchingLabels(testDeploy.Spec.Selector.MatchLabels))).To(gomega.Succeed())
+					for _, pod := range pods.Items {
+						g.Expect(pod.Spec.SchedulingGates).Should(gomega.BeComparableTo([]corev1.PodSchedulingGate{
+							{Name: "kueue.x-k8s.io/admission"},
+						}))
+						g.Expect(pod.ObjectMeta.Labels).To(gomega.BeComparableTo(map[string]string{constants.ManagedByKueueLabel: "true"}))
+						g.Expect(pod.Finalizers).Should(gomega.ContainElement(constants.ManagedByKueueLabel))
+					}
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+
+		ginkgo.It("should not suspend the pods created by a deployment in the kube-system namespace", func() {
+			var testDeploy *appsv1.Deployment
+			ginkgo.By("creating a deployment without a queue name in the kube-system namespace", func() {
+				testDeploy = testingdeploy.MakeDeployment("test-deploy", "kube-system").
+					Image(util.E2eTestAgnHostImage, util.BehaviorWaitForDeletion).
+					Request(corev1.ResourceCPU, "1").
+					Request(corev1.ResourceMemory, "2Gi").
+					Replicas(2).
+					Obj()
+				gomega.Expect(k8sClient.Create(ctx, testDeploy)).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying that the pods of the deployment are running", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					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.Status.Phase).Should(gomega.Equal(corev1.PodRunning))
+					}
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verifying that deleting the deployment removes the pods", func() {
+				pods := &corev1.PodList{}
+				gomega.Expect(k8sClient.Delete(ctx, testDeploy)).Should(gomega.Succeed())
+				gomega.Eventually(func(g gomega.Gomega) {
+					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.BeEmpty())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
 	})
 })
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index c6af19dc7..6f91d4b0e 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -605,7 +605,7 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
 			})
 
-			ginkgo.By("Finishing the jobset pods", func() {
+			ginkgo.By("Finishing the wrapped job's pods", func() {
 				util.WaitForActivePodsAndTerminate(ctx, k8sWorker1Client, worker1RestClient, worker1Cfg, aw.Namespace, 2, 0)
 			})
 
diff --git a/test/e2e/singlecluster/leaderworkerset_test.go b/test/e2e/singlecluster/leaderworkerset_test.go
index 9d9e6763b..89452431c 100644
--- a/test/e2e/singlecluster/leaderworkerset_test.go
+++ b/test/e2e/singlecluster/leaderworkerset_test.go
@@ -422,7 +422,7 @@ var _ = ginkgo.Describe("LeaderWorkerSet integration", func() {
 				})
 
 				ginkgo.By("Check workload for group 2 is deleted", func() {
-					util.ExpectObjectToBeDeleted(ctx, k8sClient, createdWorkload2, false)
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload2, false, util.LongTimeout)
 				})
 
 				ginkgo.By("Delete the LeaderWorkerSet", func() {
diff --git a/test/integration/multikueue/multikueue_test.go b/test/integration/multikueue/multikueue_test.go
index 971e4cce4..0421b6ba5 100644
--- a/test/integration/multikueue/multikueue_test.go
+++ b/test/integration/multikueue/multikueue_test.go
@@ -184,9 +184,7 @@ var _ = ginkgo.Describe("Multikueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 			acKey := client.ObjectKeyFromObject(multikueueAC)
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, acKey, &updatedAc)).To(gomega.Succeed())
-				cond := apimeta.FindStatusCondition(updatedAc.Status.Conditions, kueue.AdmissionCheckActive)
-				g.Expect(cond).NotTo(gomega.BeNil())
-				g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue), "Reason: %s, Message: %q", cond.Reason, cond.Message)
+				g.Expect(updatedAc.Status.Conditions).To(utiltesting.HaveConditionStatusTrue(kueue.AdmissionCheckActive))
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
@@ -1825,9 +1823,7 @@ var _ = ginkgo.Describe("Multikueue no GC", ginkgo.Ordered, ginkgo.ContinueOnFai
 			acKey := client.ObjectKeyFromObject(multikueueAC)
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, acKey, &updatedAc)).To(gomega.Succeed())
-				cond := apimeta.FindStatusCondition(updatedAc.Status.Conditions, kueue.AdmissionCheckActive)
-				g.Expect(cond).NotTo(gomega.BeNil())
-				g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue), "Reason: %s, Message: %q", cond.Reason, cond.Message)
+				g.Expect(updatedAc.Status.Conditions).To(utiltesting.HaveConditionStatusTrue(kueue.AdmissionCheckActive))
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
@@ -2056,9 +2052,7 @@ var _ = ginkgo.Describe("Multikueue when not all integrations are enabled", gink
 			acKey := client.ObjectKeyFromObject(multikueueAC)
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, acKey, &updatedAc)).To(gomega.Succeed())
-				cond := apimeta.FindStatusCondition(updatedAc.Status.Conditions, kueue.AdmissionCheckActive)
-				g.Expect(cond).NotTo(gomega.BeNil())
-				g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue), "Reason: %s, Message: %q", cond.Reason, cond.Message)
+				g.Expect(updatedAc.Status.Conditions).To(utiltesting.HaveConditionStatusTrue(kueue.AdmissionCheckActive))
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 
diff --git a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
index e2747c0b6..721022d78 100644
--- a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
+++ b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
@@ -1656,9 +1656,7 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
-					cond := apimeta.FindStatusCondition(updatedWl.Status.Conditions, kueue.WorkloadRequeued)
-					g.Expect(cond).ToNot(gomega.BeNil())
-					g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue))
+					g.Expect(updatedWl.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadRequeued))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -1771,9 +1769,7 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
-					cond := apimeta.FindStatusCondition(updatedWl.Status.Conditions, kueue.WorkloadRequeued)
-					g.Expect(cond).ToNot(gomega.BeNil())
-					g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue))
+					g.Expect(updatedWl.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadRequeued))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -1907,9 +1903,7 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
-					cond := apimeta.FindStatusCondition(updatedWl.Status.Conditions, kueue.WorkloadRequeued)
-					g.Expect(cond).ToNot(gomega.BeNil())
-					g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue))
+					g.Expect(updatedWl.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.WorkloadRequeued))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
diff --git a/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go b/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
index d377f4596..b2061ac8f 100644
--- a/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
@@ -691,7 +691,7 @@ var _ = ginkgo.Describe("AppWrapper controller when waitForPodsReady enabled", g
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -701,7 +701,7 @@ var _ = ginkgo.Describe("AppWrapper controller when waitForPodsReady enabled", g
 					{
 						Type:               string(awv1beta2.PodsReady),
 						Status:             metav1.ConditionTrue,
-						Reason:             "PodsReady",
+						Reason:             kueue.WorkloadStarted,
 						LastTransitionTime: metav1.NewTime(time.Now()),
 					},
 				},
@@ -709,15 +709,15 @@ var _ = ginkgo.Describe("AppWrapper controller when waitForPodsReady enabled", g
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running AppWrapper; PodsReady=False before", podsReadyTestSpec{
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			appWrapperStatus: awv1beta2.AppWrapperStatus{
@@ -725,7 +725,7 @@ var _ = ginkgo.Describe("AppWrapper controller when waitForPodsReady enabled", g
 					{
 						Type:               string(awv1beta2.PodsReady),
 						Status:             metav1.ConditionTrue,
-						Reason:             "PodsReady",
+						Reason:             kueue.WorkloadStarted,
 						LastTransitionTime: metav1.NewTime(time.Now()),
 					},
 				},
@@ -733,8 +733,8 @@ var _ = ginkgo.Describe("AppWrapper controller when waitForPodsReady enabled", g
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("AppWrapper suspended; PodsReady=True before", podsReadyTestSpec{
@@ -743,7 +743,7 @@ var _ = ginkgo.Describe("AppWrapper controller when waitForPodsReady enabled", g
 					{
 						Type:               string(awv1beta2.PodsReady),
 						Status:             metav1.ConditionTrue,
-						Reason:             "PodsReady",
+						Reason:             kueue.WorkloadStarted,
 						LastTransitionTime: metav1.NewTime(time.Now()),
 					},
 				},
@@ -751,14 +751,14 @@ var _ = ginkgo.Describe("AppWrapper controller when waitForPodsReady enabled", g
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			suspended: true,
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
index a222e25ce..bb93d96d2 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -19,6 +19,7 @@ package job
 import (
 	"fmt"
 	"maps"
+	"time"
 
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/onsi/ginkgo/v2"
@@ -1028,8 +1029,16 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 					// also updated due to setting of the job status.
 					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
 					g.Expect(util.SetQuotaReservation(ctx, k8sClient, createdWorkload, nil)).Should(gomega.Succeed())
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
+					g.Expect(createdWorkload.Status.Conditions).To(gomega.ContainElements(
+						gomega.BeComparableTo(metav1.Condition{
+							Type:    kueue.WorkloadAdmitted,
+							Status:  metav1.ConditionFalse,
+							Reason:  "NoReservation",
+							Message: "The workload has no reservation",
+						}, util.IgnoreConditionTimestampsAndObservedGeneration),
+					))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
-				util.SyncAdmittedConditionForWorkloads(ctx, k8sClient, createdWorkload)
 			}
 
 			ginkgo.By("Verify the PodsReady condition is added")
@@ -1043,7 +1052,7 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -1055,7 +1064,7 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -1066,7 +1075,7 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -1078,8 +1087,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("One pod ready, one terminating succeeded", podsReadyTestSpec{
@@ -1093,8 +1102,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("One pod ready, one succeeded", podsReadyTestSpec{
@@ -1106,8 +1115,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("One pod succeeded, one terminating succeeded", podsReadyTestSpec{
@@ -1120,8 +1129,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("All pods terminating succeeded", podsReadyTestSpec{
@@ -1133,8 +1142,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("All pods are succeeded", podsReadyTestSpec{
@@ -1145,15 +1154,15 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("All pods are succeeded; PodsReady=False before", podsReadyTestSpec{
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			jobStatus: batchv1.JobStatus{
@@ -1163,8 +1172,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("One ready pod, one failed; PodsReady=True before", podsReadyTestSpec{
@@ -1175,8 +1184,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			jobStatus: batchv1.JobStatus{
 				Active: 1,
@@ -1185,9 +1194,9 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			},
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
-				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Status:  metav1.ConditionFalse,
+				Reason:  kueue.WorkloadWaitForRecovery,
+				Message: "At least one pod has failed, waiting for recovery",
 			},
 		}),
 		ginkgo.Entry("Job suspended without ready pods; but PodsReady=True before", podsReadyTestSpec{
@@ -1198,8 +1207,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			jobStatus: batchv1.JobStatus{
 				Failed: 2,
@@ -1208,7 +1217,7 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -1220,8 +1229,8 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			jobStatus: batchv1.JobStatus{
 				Active: 2,
@@ -1231,7 +1240,7 @@ var _ = ginkgo.Describe("When waitForPodsReady enabled", ginkgo.Ordered, ginkgo.
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -2066,26 +2075,29 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 	})
 })
 
-var _ = ginkgo.Describe("Job controller interacting with Workload controller when waitForPodsReady with requeuing strategy is enabled", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+var _ = ginkgo.Describe("Job controller interacting with Workload controller when waitForPodsReady is enabled", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
 	var (
-		backoffBaseSeconds int32
-		backoffLimitCount  *int32
-		ns                 *corev1.Namespace
-		fl                 *kueue.ResourceFlavor
-		cq                 *kueue.ClusterQueue
-		lq                 *kueue.LocalQueue
+		backoffBaseSeconds              int32
+		backoffLimitCount               *int32
+		waitForPodsReadyTimeout         *metav1.Duration
+		waitForPodsReadyRecoveryTimeout *metav1.Duration
+		ns                              *corev1.Namespace
+		fl                              *kueue.ResourceFlavor
+		cq                              *kueue.ClusterQueue
+		lq                              *kueue.LocalQueue
 	)
 
 	ginkgo.JustBeforeEach(func() {
 		waitForPodsReady := &configapi.WaitForPodsReady{
 			Enable:         true,
 			BlockAdmission: ptr.To(true),
-			Timeout:        &metav1.Duration{Duration: util.TinyTimeout},
+			Timeout:        waitForPodsReadyTimeout,
 			RequeuingStrategy: &configapi.RequeuingStrategy{
 				Timestamp:          ptr.To(configapi.EvictionTimestamp),
 				BackoffBaseSeconds: ptr.To(backoffBaseSeconds),
 				BackoffLimitCount:  backoffLimitCount,
 			},
+			RecoveryTimeout: waitForPodsReadyRecoveryTimeout,
 		}
 		fwk.StartManager(ctx, cfg, managerAndControllersSetup(
 			false,
@@ -2115,12 +2127,13 @@ var _ = ginkgo.Describe("Job controller interacting with Workload controller whe
 		fwk.StopManager(ctx)
 	})
 
-	ginkgo.When("long backoffBaseSeconds", func() {
+	ginkgo.When("long backoffBaseSeconds and tiny waitForPodsReady.timeout", func() {
 		ginkgo.BeforeEach(func() {
 			backoffBaseSeconds = 10
+			waitForPodsReadyTimeout = &metav1.Duration{Duration: util.TinyTimeout}
 		})
 
-		ginkgo.It("should evict workload due pods ready timeout", func() {
+		ginkgo.It("should evict workload due waitForPodsReady.timeout", func() {
 			ginkgo.By("creating job")
 			job := testingjob.MakeJob("job", ns.Name).Queue(lq.Name).Request(corev1.ResourceCPU, "2").Obj()
 			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
@@ -2141,7 +2154,7 @@ var _ = ginkgo.Describe("Job controller interacting with Workload controller whe
 					gomega.BeComparableTo(metav1.Condition{
 						Type:    kueue.WorkloadPodsReady,
 						Status:  metav1.ConditionFalse,
-						Reason:  kueue.WorkloadPodsReady,
+						Reason:  kueue.WorkloadWaitForStart,
 						Message: "Not all pods are ready or succeeded",
 					}, util.IgnoreConditionTimestampsAndObservedGeneration),
 					gomega.BeComparableTo(metav1.Condition{
@@ -2173,10 +2186,239 @@ var _ = ginkgo.Describe("Job controller interacting with Workload controller whe
 		})
 	})
 
-	ginkgo.When("short backoffBaseSeconds", func() {
+	ginkgo.When("Job is running and a pod fails without .recoveryTimeout configured", func() {
+		ginkgo.BeforeEach(func() {
+			waitForPodsReadyTimeout = &metav1.Duration{Duration: 5 * time.Minute}
+			waitForPodsReadyRecoveryTimeout = nil
+		})
+
+		ginkgo.It("shouldn't evict workload due waitForPodsReady.recoveryTimeout", func() {
+			ginkgo.By("creating job")
+			job := testingjob.MakeJob("job", ns.Name).Queue(lq.Name).Request(corev1.ResourceCPU, "2").Obj()
+			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
+			jobKey := client.ObjectKeyFromObject(job)
+
+			wl := &kueue.Workload{}
+			wlKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: job.Namespace}
+
+			ginkgo.By("setting quota reservation")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(ctx, k8sClient, wl, testing.MakeAdmission(cq.Name).Obj())).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("setting all job's pods to be ready")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, jobKey, job)).Should(gomega.Succeed())
+				job.Status.Active = 1
+				job.Status.Ready = ptr.To[int32](1)
+				gomega.Expect(k8sClient.Status().Update(ctx, job)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("ensuring workload has PodsReady=True condition")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Status.Conditions).Should(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadPodsReady,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadStarted,
+						Message: "All pods reached readiness and the workload is running",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration)))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("failing one pod")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, jobKey, job)).Should(gomega.Succeed())
+				job.Status.Active = 0
+				job.Status.Ready = ptr.To[int32](0)
+				job.Status.Failed = 1
+				gomega.Expect(k8sClient.Status().Update(ctx, job)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("checking the workload is still admitted")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl)
+				util.ExpectReservingActiveWorkloadsMetric(cq, 1)
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
+
+	ginkgo.When("A tiny recoveryTimeout is configured and a pod fails", func() {
+		ginkgo.BeforeEach(func() {
+			waitForPodsReadyTimeout = &metav1.Duration{Duration: 5 * time.Minute}
+			waitForPodsReadyRecoveryTimeout = &metav1.Duration{Duration: util.TinyTimeout}
+		})
+
+		ginkgo.It("should evict workload due waitForPodsReady.recoveryTimeout", func() {
+			ginkgo.By("creating job")
+			job := testingjob.MakeJob("job", ns.Name).Queue(lq.Name).Request(corev1.ResourceCPU, "2").Obj()
+			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
+			jobKey := client.ObjectKeyFromObject(job)
+
+			wl := &kueue.Workload{}
+			wlKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: job.Namespace}
+
+			ginkgo.By("setting quota reservation")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(ctx, k8sClient, wl, testing.MakeAdmission(cq.Name).Obj())).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("setting all job's pods to be ready")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, jobKey, job)).Should(gomega.Succeed())
+				job.Status.Active = 1
+				job.Status.Ready = ptr.To[int32](1)
+				gomega.Expect(k8sClient.Status().Update(ctx, job)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("ensuring workload has PodsReady=True condition")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Status.Conditions).Should(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadPodsReady,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadStarted,
+						Message: "All pods reached readiness and the workload is running",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration)))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("failing one pod")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, jobKey, job)).Should(gomega.Succeed())
+				job.Status.Active = 0
+				job.Status.Ready = ptr.To[int32](0)
+				job.Status.Failed = 1
+				gomega.Expect(k8sClient.Status().Update(ctx, job)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("checking the workload is evicted")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Status.Conditions).To(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadPodsReady,
+						Status:  metav1.ConditionFalse,
+						Reason:  kueue.WorkloadWaitForStart,
+						Message: "Not all pods are ready or succeeded",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration),
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadEvicted,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadEvictedByPodsReadyTimeout,
+						Message: fmt.Sprintf("Exceeded the PodsReady timeout %s", wlKey.String()),
+					}, util.IgnoreConditionTimestampsAndObservedGeneration),
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadAdmitted,
+						Status:  metav1.ConditionFalse,
+						Reason:  "NoReservation",
+						Message: "The workload has no reservation",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration),
+				))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
+
+	ginkgo.When("A long recoveryTimeout is configured and a pod fails", func() {
+		ginkgo.BeforeEach(func() {
+			waitForPodsReadyTimeout = &metav1.Duration{Duration: 5 * time.Minute}
+			waitForPodsReadyRecoveryTimeout = &metav1.Duration{Duration: util.LongTimeout}
+		})
+
+		ginkgo.It("should wait for .recoveryTimeout for a workload to recover", func() {
+			ginkgo.By("creating job")
+			job := testingjob.MakeJob("job", ns.Name).Queue(lq.Name).Request(corev1.ResourceCPU, "2").Obj()
+			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
+			jobKey := client.ObjectKeyFromObject(job)
+
+			wl := &kueue.Workload{}
+			wlKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: job.Namespace}
+
+			ginkgo.By("setting quota reservation")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(util.SetQuotaReservation(ctx, k8sClient, wl, testing.MakeAdmission(cq.Name).Obj())).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("setting all job's pods to be ready")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, jobKey, job)).Should(gomega.Succeed())
+				job.Status.Active = 1
+				job.Status.Ready = ptr.To[int32](1)
+				gomega.Expect(k8sClient.Status().Update(ctx, job)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("ensuring workload has PodsReady=True condition")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Status.Conditions).Should(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadPodsReady,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadStarted,
+						Message: "All pods reached readiness and the workload is running",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration)))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("failing one pod")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, jobKey, job)).Should(gomega.Succeed())
+				job.Status.Active = 0
+				job.Status.Ready = ptr.To[int32](0)
+				job.Status.Failed = 1
+				gomega.Expect(k8sClient.Status().Update(ctx, job)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("ensuring workload has PodsReady=False condition")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Status.Conditions).Should(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadPodsReady,
+						Status:  metav1.ConditionFalse,
+						Reason:  kueue.WorkloadWaitForRecovery,
+						Message: "At least one pod has failed, waiting for recovery",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration)))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("recovering the pod")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, jobKey, job)).Should(gomega.Succeed())
+				job.Status.Active = 1
+				job.Status.Ready = ptr.To[int32](1)
+				job.Status.Failed = 1
+				gomega.Expect(k8sClient.Status().Update(ctx, job)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("ensuring workload has PodsReady=True condition")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Status.Conditions).Should(gomega.ContainElements(
+					gomega.BeComparableTo(metav1.Condition{
+						Type:    kueue.WorkloadPodsReady,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadRecovered,
+						Message: "All pods reached readiness and the workload is running",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration)))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("checking the workload is still admitted")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).Should(gomega.Succeed())
+				util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl)
+				util.ExpectReservingActiveWorkloadsMetric(cq, 1)
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
+
+	ginkgo.When("short backoffBaseSeconds and tiny .timeout", func() {
 		ginkgo.BeforeEach(func() {
 			backoffBaseSeconds = 1
 			backoffLimitCount = ptr.To[int32](1)
+			waitForPodsReadyTimeout = &metav1.Duration{Duration: util.TinyTimeout}
 		})
 
 		ginkgo.It("should re-queue a workload evicted due to PodsReady timeout after the backoff elapses", func() {
@@ -2209,7 +2451,7 @@ var _ = ginkgo.Describe("Job controller interacting with Workload controller whe
 					gomega.BeComparableTo(metav1.Condition{
 						Type:    kueue.WorkloadPodsReady,
 						Status:  metav1.ConditionFalse,
-						Reason:  kueue.WorkloadPodsReady,
+						Reason:  kueue.WorkloadWaitForStart,
 						Message: "Not all pods are ready or succeeded",
 					}, util.IgnoreConditionTimestampsAndObservedGeneration),
 					gomega.BeComparableTo(metav1.Condition{
@@ -2256,7 +2498,7 @@ var _ = ginkgo.Describe("Job controller interacting with Workload controller whe
 					gomega.BeComparableTo(metav1.Condition{
 						Type:    kueue.WorkloadPodsReady,
 						Status:  metav1.ConditionFalse,
-						Reason:  kueue.WorkloadPodsReady,
+						Reason:  kueue.WorkloadWaitForStart,
 						Message: "Not all pods are ready or succeeded",
 					}, util.IgnoreConditionTimestampsAndObservedGeneration),
 					gomega.BeComparableTo(metav1.Condition{
diff --git a/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go b/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go
index f89fac630..c33efcf2e 100644
--- a/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/jobset/jobset_controller_test.go
@@ -882,7 +882,7 @@ var _ = ginkgo.Describe("JobSet controller when waitForPodsReady enabled", ginkg
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -904,15 +904,15 @@ var _ = ginkgo.Describe("JobSet controller when waitForPodsReady enabled", ginkg
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running JobSet; PodsReady=False before", podsReadyTestSpec{
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			jobSetStatus: jobsetapi.JobSetStatus{
@@ -932,8 +932,8 @@ var _ = ginkgo.Describe("JobSet controller when waitForPodsReady enabled", ginkg
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("JobSet suspended; PodsReady=True before", podsReadyTestSpec{
@@ -954,14 +954,14 @@ var _ = ginkgo.Describe("JobSet controller when waitForPodsReady enabled", ginkg
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			suspended: true,
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go b/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go
index 0e557d0cf..4dbf34d14 100644
--- a/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/mpijob/mpijob_controller_test.go
@@ -709,7 +709,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -726,15 +726,15 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running MPIJob; PodsReady=False before", podsReadyTestSpec{
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			jobStatus: kfmpi.JobStatus{
@@ -749,8 +749,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Job suspended; PodsReady=True before", podsReadyTestSpec{
@@ -766,8 +766,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			jobStatus: kfmpi.JobStatus{
 				Conditions: []kfmpi.JobCondition{
@@ -782,7 +782,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go b/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go
index 20fff921f..46b9e5b89 100644
--- a/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/paddlejob/paddlejob_controller_test.go
@@ -153,7 +153,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -170,15 +170,15 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running PaddleJob; PodsReady=False before", kftesting.PodsReadyTestSpec{
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			JobStatus: kftraining.JobStatus{
@@ -193,8 +193,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Job suspended; PodsReady=True before", kftesting.PodsReadyTestSpec{
@@ -210,8 +210,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			JobStatus: kftraining.JobStatus{
 				Conditions: []kftraining.JobCondition{
@@ -226,7 +226,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
index 4380662b9..cf05126d7 100644
--- a/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/pod/pod_controller_test.go
@@ -26,7 +26,6 @@ import (
 	"github.com/onsi/gomega"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
-	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -2028,7 +2027,7 @@ var _ = ginkgo.Describe("Pod controller interacting with Workload controller whe
 					gomega.BeComparableTo(metav1.Condition{
 						Type:    kueue.WorkloadPodsReady,
 						Status:  metav1.ConditionFalse,
-						Reason:  kueue.WorkloadPodsReady,
+						Reason:  kueue.WorkloadWaitForStart,
 						Message: "Not all pods are ready or succeeded",
 					}, util.IgnoreConditionTimestampsAndObservedGeneration),
 				))
@@ -2055,7 +2054,7 @@ var _ = ginkgo.Describe("Pod controller interacting with Workload controller whe
 					gomega.BeComparableTo(metav1.Condition{
 						Type:    kueue.WorkloadPodsReady,
 						Status:  metav1.ConditionFalse,
-						Reason:  kueue.WorkloadPodsReady,
+						Reason:  kueue.WorkloadWaitForStart,
 						Message: "Not all pods are ready or succeeded",
 					}, util.IgnoreConditionTimestampsAndObservedGeneration),
 					gomega.BeComparableTo(metav1.Condition{
@@ -2105,7 +2104,7 @@ var _ = ginkgo.Describe("Pod controller interacting with Workload controller whe
 					gomega.BeComparableTo(metav1.Condition{
 						Type:    kueue.WorkloadPodsReady,
 						Status:  metav1.ConditionFalse,
-						Reason:  kueue.WorkloadPodsReady,
+						Reason:  kueue.WorkloadWaitForStart,
 						Message: "Not all pods are ready or succeeded",
 					}, util.IgnoreConditionTimestampsAndObservedGeneration),
 					gomega.BeComparableTo(metav1.Condition{
@@ -2133,7 +2132,7 @@ var _ = ginkgo.Describe("Pod controller interacting with Workload controller whe
 						Message: "The workload is deactivated due to exceeding the maximum number of re-queuing retries",
 					}, util.IgnoreConditionTimestampsAndObservedGeneration),
 				))
-				g.Expect(apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadDeactivationTarget)).Should(gomega.BeNil())
+				g.Expect(wl.Status.Conditions).ShouldNot(testing.HaveCondition(kueue.WorkloadDeactivationTarget))
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 	})
diff --git a/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go b/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go
index 5ec6f95ce..7e5d030a8 100644
--- a/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/pytorchjob/pytorchjob_controller_test.go
@@ -386,7 +386,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -403,15 +403,15 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running PyTorchJob; PodsReady=False before", kftesting.PodsReadyTestSpec{
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			JobStatus: kftraining.JobStatus{
@@ -426,8 +426,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Job suspended; PodsReady=True before", kftesting.PodsReadyTestSpec{
@@ -443,8 +443,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			JobStatus: kftraining.JobStatus{
 				Conditions: []kftraining.JobCondition{
@@ -459,7 +459,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go b/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go
index 2ecd47bbd..63a5c3f3a 100644
--- a/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/raycluster/raycluster_controller_test.go
@@ -430,7 +430,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -441,8 +441,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 
@@ -450,7 +450,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			jobStatus: rayv1.RayClusterStatus{
@@ -460,8 +460,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Job suspended; PodsReady=True before", podsReadyTestSpec{
@@ -471,8 +471,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			jobStatus: rayv1.RayClusterStatus{
 				State: rayv1.Ready,
@@ -481,7 +481,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/rayjob/rayjob_controller_test.go b/test/integration/singlecluster/controller/jobs/rayjob/rayjob_controller_test.go
index a079010b8..61a1a6564 100644
--- a/test/integration/singlecluster/controller/jobs/rayjob/rayjob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/rayjob/rayjob_controller_test.go
@@ -461,7 +461,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -475,15 +475,15 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running RayJob; PodsReady=False before", podsReadyTestSpec{
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			jobStatus: rayv1.RayJobStatus{
@@ -495,8 +495,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Job suspended; PodsReady=True before", podsReadyTestSpec{
@@ -509,8 +509,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			beforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			jobStatus: rayv1.RayJobStatus{
 				JobDeploymentStatus: rayv1.JobDeploymentStatusSuspended,
@@ -519,7 +519,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", ginkgo.O
 			wantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go b/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go
index dd4340587..eb9401190 100644
--- a/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/tfjob/tfjob_controller_test.go
@@ -162,7 +162,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -179,15 +179,15 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running TFJob; PodsReady=False before", kftesting.PodsReadyTestSpec{
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			JobStatus: kftraining.JobStatus{
@@ -202,8 +202,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Job suspended; PodsReady=True before", kftesting.PodsReadyTestSpec{
@@ -219,8 +219,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			JobStatus: kftraining.JobStatus{
 				Conditions: []kftraining.JobCondition{
@@ -235,7 +235,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go b/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go
index ffdede17d..3fd5539d2 100644
--- a/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/xgboostjob/xgboostjob_controller_test.go
@@ -152,7 +152,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
@@ -169,15 +169,15 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Running XGBoostJob; PodsReady=False before", kftesting.PodsReadyTestSpec{
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 			JobStatus: kftraining.JobStatus{
@@ -192,8 +192,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 		}),
 		ginkgo.Entry("Job suspended; PodsReady=True before", kftesting.PodsReadyTestSpec{
@@ -209,8 +209,8 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			BeforeCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionTrue,
-				Reason:  "PodsReady",
-				Message: "All pods were ready or succeeded since the workload admission",
+				Reason:  kueue.WorkloadStarted,
+				Message: "All pods reached readiness and the workload is running",
 			},
 			JobStatus: kftraining.JobStatus{
 				Conditions: []kftraining.JobCondition{
@@ -225,7 +225,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", framewor
 			WantCondition: &metav1.Condition{
 				Type:    kueue.WorkloadPodsReady,
 				Status:  metav1.ConditionFalse,
-				Reason:  "PodsReady",
+				Reason:  kueue.WorkloadWaitForStart,
 				Message: "Not all pods are ready or succeeded",
 			},
 		}),
diff --git a/test/integration/singlecluster/scheduler/podsready/scheduler_test.go b/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
index 58db1ee80..c1f58aa7e 100644
--- a/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
+++ b/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
@@ -140,7 +140,7 @@ var _ = ginkgo.Describe("SchedulerWithWaitForPodsReady", func() {
 			util.ExpectWorkloadsToHaveQuotaReservation(ctx, k8sClient, prodClusterQ.Name, prodWl)
 			util.ExpectWorkloadsToBeWaiting(ctx, k8sClient, devWl)
 
-			ginkgo.By("update the first workload to be in the PodsReady condition and verify the second workload is admitted")
+			ginkgo.By("update the first workload with PodsReady=True, reason=PodsReady condition and verify the second workload is admitted")
 			gomega.Eventually(func(g gomega.Gomega) {
 				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(prodWl), prodWl)).Should(gomega.Succeed())
 				apimeta.SetStatusCondition(&prodWl.Status.Conditions, metav1.Condition{
@@ -153,6 +153,28 @@ var _ = ginkgo.Describe("SchedulerWithWaitForPodsReady", func() {
 			util.ExpectWorkloadsToHaveQuotaReservation(ctx, k8sClient, devClusterQ.Name, devWl)
 		})
 
+		ginkgo.It("Should unblock admission of new workloads in other ClusterQueues once the admitted workload exceeds timeout", func() {
+			ginkgo.By("checking the first prod workload gets admitted while the second is waiting")
+			prodWl := testing.MakeWorkload("prod-wl", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "2").Obj()
+			gomega.Expect(k8sClient.Create(ctx, prodWl)).Should(gomega.Succeed())
+			devWl := testing.MakeWorkload("dev-wl", ns.Name).Queue(devQueue.Name).Request(corev1.ResourceCPU, "2").Obj()
+			gomega.Expect(k8sClient.Create(ctx, devWl)).Should(gomega.Succeed())
+			util.ExpectWorkloadsToHaveQuotaReservation(ctx, k8sClient, prodClusterQ.Name, prodWl)
+			util.ExpectWorkloadsToBeWaiting(ctx, k8sClient, devWl)
+
+			ginkgo.By("update the first workload with PodsReady=True, reason=Started condition and verify the second workload is admitted")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(prodWl), prodWl)).Should(gomega.Succeed())
+				apimeta.SetStatusCondition(&prodWl.Status.Conditions, metav1.Condition{
+					Type:   kueue.WorkloadPodsReady,
+					Status: metav1.ConditionTrue,
+					Reason: kueue.WorkloadStarted,
+				})
+				g.Expect(k8sClient.Status().Update(ctx, prodWl)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			util.ExpectWorkloadsToHaveQuotaReservation(ctx, k8sClient, devClusterQ.Name, devWl)
+		})
+
 		ginkgo.It("Should unblock admission of new workloads once the admitted workload is deleted", func() {
 			ginkgo.By("checking the first prod workload gets admitted while the second is waiting")
 			prodWl := testing.MakeWorkload("prod-wl", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "2").Obj()
diff --git a/test/util/util.go b/test/util/util.go
index c6908fdc4..33f6a187e 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -860,9 +860,7 @@ func ExpectClusterQueuesToBeActive(ctx context.Context, c client.Client, cqs ...
 		readCq := &kueue.ClusterQueue{}
 		for _, cq := range cqs {
 			g.Expect(c.Get(ctx, client.ObjectKeyFromObject(cq), readCq)).To(gomega.Succeed())
-			cond := apimeta.FindStatusCondition(readCq.Status.Conditions, kueue.ClusterQueueActive)
-			g.Expect(cond).NotTo(gomega.BeNil(), "no %q condition found in %q cq status", kueue.ClusterQueueActive, cq.Name)
-			g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue), "%q is not active status: %q message: %q", cq.Name, cond.Status, cond.Message)
+			g.Expect(readCq.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.ClusterQueueActive))
 		}
 	}, Timeout, Interval).Should(gomega.Succeed())
 }
@@ -872,9 +870,7 @@ func ExpectLocalQueuesToBeActive(ctx context.Context, c client.Client, lqs ...*k
 		readLq := &kueue.LocalQueue{}
 		for _, lq := range lqs {
 			g.Expect(c.Get(ctx, client.ObjectKeyFromObject(lq), readLq)).To(gomega.Succeed())
-			cond := apimeta.FindStatusCondition(readLq.Status.Conditions, kueue.LocalQueueActive)
-			g.Expect(cond).NotTo(gomega.BeNil(), "no %q condition found in %q cq status", kueue.LocalQueueActive, lq.Name)
-			g.Expect(cond.Status).To(gomega.Equal(metav1.ConditionTrue), "%q is not active status: %q message: %q", lq.Name, cond.Status, cond.Message)
+			g.Expect(readLq.Status.Conditions).To(testing.HaveConditionStatusTrue(kueue.LocalQueueActive))
 		}
 	}, Timeout, Interval).Should(gomega.Succeed())
 }