Repository: kubernetes-sigs/kueue

Issue:
Add retry mechanism for AdmissionChecks in Kueue
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:
Retry mechanism for AdmissionChecks.

**Why is this needed**:
Currently there is no retry mechanism for AdmissionChecks. There is a deprecated field `admissionCheckSpec.retryDelayMinutes` which was never supported and will be remove in API v1beta2. In the AdmissionCheck API there is the `Retry`, however the workload controller evicts the Workload without retrying it. Hence, there is a lot of improvement potential for the retry mechanism.

**Completion requirements**:

This enhancement requires the following artifacts:

- [ ] Design doc
- [ ] API change
- [ ] Docs update

The artifacts should be linked in subsequent comments.


Discussion:
@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: /remove-lifecycle stale@Horiodino: /assign @PBundyra  is this issue still opened ? @Horiodino: @PBundyra  can you specify in `AdmissionCheck` API  where is  `Retry` ?@mimowo: `Retry` is one of the [states](https://github.com/kubernetes-sigs/kueue/blob/fd2519a76bc4d51b3b464d1f27525136a69b080e/apis/kueue/v1beta1/admissioncheck_types.go#L26-L30) an AdmissionCheck could be in.

We set in some scenarios, such as ProvisioningRequest Retry (see [here](https://github.com/kubernetes-sigs/kueue/blob/fd2519a76bc4d51b3b464d1f27525136a69b080e/pkg/controller/admissionchecks/provisioning/controller.go#L552). However, currently the retry mechanism is coupled to ProvisioningRequest. It means if there are some external ACs they need to retry on their own. 

The issue will require designing the API for a generic mechanism for retrying an AC. However, I believe the issue was created as one of the alternatives to retry ProvisioningRequests (see [Support for configurable ProvisioningRequest retries](https://github.com/kubernetes-sigs/kueue/issues/1353)). In other words: the actual business need of being able to retry ProvisioningRequests had these two competing approaches. Since we started to support retries at the ProvisioningRequest-level I think the relative priority of this issue has decreased. 

While still the mechanism could be useful for other AdmissionChecks, I would like to hear first in the community how much of a need there remains, before investing time into development and reviews.
@leipanhz: > `Retry` is one of the [states](https://github.com/kubernetes-sigs/kueue/blob/fd2519a76bc4d51b3b464d1f27525136a69b080e/apis/kueue/v1beta1/admissioncheck_types.go#L26-L30) an AdmissionCheck could be in.
> 
> We set in some scenarios, such as ProvisioningRequest Retry (see [here](https://github.com/kubernetes-sigs/kueue/blob/fd2519a76bc4d51b3b464d1f27525136a69b080e/pkg/controller/admissionchecks/provisioning/controller.go#L552). However, currently the retry mechanism is coupled to ProvisioningRequest. It means if there are some external ACs they need to retry on their own.
> 
> The issue will require designing the API for a generic mechanism for retrying an AC. However, I believe the issue was created as one of the alternatives to retry ProvisioningRequests (see [Support for configurable ProvisioningRequest retries](https://github.com/kubernetes-sigs/kueue/issues/1353)). In other words: the actual business need of being able to retry ProvisioningRequests had these two competing approaches. Since we started to support retries at the ProvisioningRequest-level I think the relative priority of this issue has decreased.
> 
> While still the mechanism could be useful for other AdmissionChecks, I would like to hear first in the community how much of a need there remains, before investing time into development and reviews.

Hi @mimowo , thanks for documenting this. Retry serves an important role for external admission check to work properly. Imho, the fact it only works with internal provision requests would limit Kueue's applicabilty. @mimowo: > the fact it only works with internal provision requests would limit Kueue's applicabilty.

Right, thanks for emphasizing the need for the feature to support external AdmissionChecks. I believe the retry delay still could be hacked as in ProvisioningRequest AdmissionCheck, but I'm aware it means duplicating and maintaining complex code. 

cc @mwielgus @mwysokin@dhenkel92: We are currently in the process of introducing Kueue to our infrastructure. One reason why we picked it is the ability to run custom admission checks to have more control over whether a workload can/should be run in a cluster or not.

For example: We've created a check that detects if the Kubernetes control-plane of that cluster is overloaded. In this case, the workload should be delayed until the control-plane is healthy again.

Unfortunately, without retries, Kueue will constantly reconcile the workload, which does not help the control-plane recover.
Right now, we've built a workaround for the missing capability by misusing the `.status.requeueState` in our admission checks.
However, this is not maintainable, so it would be great to have a proper solution long term.

In case you have higher priorities, we would love to help.
@mimowo: cc @mwielgus 

yes, I fully understand support for retries is required when you use external ACs, and this is also hacked in the ProvReq controller.

Regarding the priorities - yes it does not seem to be top priority now as we have fixed ProvReq, but we would welcome a contribution.

I think we will need some API for that, so we need KEP. One question is how to merry the retries at the AC level with the retries at the ProvReq level (or other controllers). @dhenkel92: Sounds great. 

> I think we will need some API for that, so we need KEP. One question is how to merry the retries at the AC level with the retries at the ProvReq level (or other controllers).

We'll start a KEP and can discuss further there. Is there anything else you think we should take into consideration?@mimowo: Not at the moment, maybe @PBundyra has some suggestions.@PBundyra: Hi @dhenkel92, thanks for the follow-up on this. You may also take a look on that slack thread https://kubernetes.slack.com/archives/C032ZE66A2X/p1729787805510569
I believe one of your colleague also tried a similar issue so maybe he could also give some piecie of advice @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@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

Solution patch:
diff --git a/apis/config/v1beta1/configuration_conversion.go b/apis/config/v1beta1/configuration_conversion.go
index d23470cf4..2482ac8e8 100644
--- a/apis/config/v1beta1/configuration_conversion.go
+++ b/apis/config/v1beta1/configuration_conversion.go
@@ -18,6 +18,7 @@ package v1beta1
 
 import (
 	conversionapi "k8s.io/apimachinery/pkg/conversion"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/conversion"
 
 	"sigs.k8s.io/kueue/apis/config/v1beta2"
@@ -64,10 +65,22 @@ func Convert_v1beta2_FairSharing_To_v1beta1_FairSharing(in *v1beta2.FairSharing,
 }
 
 func Convert_v1beta1_WaitForPodsReady_To_v1beta2_WaitForPodsReady(in *WaitForPodsReady, out *v1beta2.WaitForPodsReady, s conversionapi.Scope) error {
-	return autoConvert_v1beta1_WaitForPodsReady_To_v1beta2_WaitForPodsReady(in, out, s)
+	if err := autoConvert_v1beta1_WaitForPodsReady_To_v1beta2_WaitForPodsReady(in, out, s); err != nil {
+		return err
+	}
+	if in.BlockAdmission == nil {
+		out.BlockAdmission = ptr.To(true)
+	}
+	return nil
 }
 
 func Convert_v1beta2_WaitForPodsReady_To_v1beta1_WaitForPodsReady(in *v1beta2.WaitForPodsReady, out *WaitForPodsReady, s conversionapi.Scope) error {
 	out.Enable = true
-	return autoConvert_v1beta2_WaitForPodsReady_To_v1beta1_WaitForPodsReady(in, out, s)
+	if err := autoConvert_v1beta2_WaitForPodsReady_To_v1beta1_WaitForPodsReady(in, out, s); err != nil {
+		return err
+	}
+	if in.BlockAdmission == nil {
+		out.BlockAdmission = ptr.To(false)
+	}
+	return nil
 }
diff --git a/apis/config/v1beta2/configuration_types.go b/apis/config/v1beta2/configuration_types.go
index 28370b1e5..8c18113cc 100644
--- a/apis/config/v1beta2/configuration_types.go
+++ b/apis/config/v1beta2/configuration_types.go
@@ -222,10 +222,9 @@ type WaitForPodsReady struct {
 	// +optional
 	Timeout *metav1.Duration `json:"timeout,omitempty"`
 
-	// BlockAdmission when true, cluster queue will block admissions for all
+	// BlockAdmission when true, the cluster queue will block admissions for all
 	// subsequent jobs until the jobs reach the PodsReady=true condition.
-	// This setting is only honored when `Enable` is set to true.
-	// Defaults to true.
+	// Defaults to false.
 	// +optional
 	BlockAdmission *bool `json:"blockAdmission,omitempty"`
 
@@ -238,8 +237,7 @@ type WaitForPodsReady struct {
 	// Such a transition may happen when a Pod failed and the replacement Pod
 	// is awaited to be scheduled.
 	// After exceeding the timeout the corresponding job gets suspended again
-	// and requeued after the backoff delay. The timeout is enforced only if waitForPodsReady.enable=true.
-	// If not set, there is no timeout.
+	// and requeued after the backoff delay.
 	// +optional
 	RecoveryTimeout *metav1.Duration `json:"recoveryTimeout,omitempty"`
 }
diff --git a/apis/config/v1beta2/defaults.go b/apis/config/v1beta2/defaults.go
index ca6dadb09..60a221ba5 100644
--- a/apis/config/v1beta2/defaults.go
+++ b/apis/config/v1beta2/defaults.go
@@ -94,7 +94,7 @@ func SetDefaults_Configuration(cfg *Configuration) {
 
 	if cfg.WaitForPodsReady != nil {
 		cfg.WaitForPodsReady.Timeout = cmp.Or(cfg.WaitForPodsReady.Timeout, &metav1.Duration{Duration: defaultPodsReadyTimeout})
-		cfg.WaitForPodsReady.BlockAdmission = cmp.Or(cfg.WaitForPodsReady.BlockAdmission, ptr.To(true))
+		cfg.WaitForPodsReady.BlockAdmission = cmp.Or(cfg.WaitForPodsReady.BlockAdmission, ptr.To(false))
 		cfg.WaitForPodsReady.RequeuingStrategy = cmp.Or(cfg.WaitForPodsReady.RequeuingStrategy, &RequeuingStrategy{})
 		cfg.WaitForPodsReady.RequeuingStrategy.Timestamp = cmp.Or(cfg.WaitForPodsReady.RequeuingStrategy.Timestamp, ptr.To(EvictionTimestamp))
 		cfg.WaitForPodsReady.RequeuingStrategy.BackoffBaseSeconds = cmp.Or(cfg.WaitForPodsReady.RequeuingStrategy.BackoffBaseSeconds, ptr.To[int32](DefaultRequeuingBackoffBaseSeconds))
diff --git a/apis/kueue/v1beta1/workload_types.go b/apis/kueue/v1beta1/workload_types.go
index 86ab44baa..2e017f0af 100644
--- a/apis/kueue/v1beta1/workload_types.go
+++ b/apis/kueue/v1beta1/workload_types.go
@@ -542,7 +542,24 @@ type AdmissionCheckState struct {
 	// +kubebuilder:validation:Required
 	// +kubebuilder:validation:MaxLength=32768
 	Message string `json:"message" protobuf:"bytes,6,opt,name=message"`
-
+	// requeueAfterSeconds indicates how long to wait at least before
+	// retrying to admit the workload.
+	// The admission check controllers can set this field when State=Retry
+	// to implement delays between retry attempts.
+	//
+	// If nil when State=Retry, Kueue will retry immediately.
+	// If set, Kueue will add the workload back to the queue after
+	//   lastTransitionTime + RequeueAfterSeconds is over.
+	//
+	// +optional
+	// +kubebuilder:validation:Minimum=0
+	RequeueAfterSeconds *int32 `json:"requeueAfterSeconds,omitempty"`
+	// retryCount tracks retry attempts for this admission check.
+	// Kueue automatically increments the counter whenever the
+	// state transitions to Retry.
+	// +optional
+	// +kubebuilder:validation:Minimum=0
+	RetryCount *int32 `json:"retryCount,omitempty"`
 	// podSetUpdates contains a list of pod set modifications suggested by AdmissionChecks.
 	// +optional
 	// +listType=atomic
diff --git a/apis/kueue/v1beta1/zz_generated.conversion.go b/apis/kueue/v1beta1/zz_generated.conversion.go
index e6a8c0db7..73c1922da 100644
--- a/apis/kueue/v1beta1/zz_generated.conversion.go
+++ b/apis/kueue/v1beta1/zz_generated.conversion.go
@@ -944,6 +944,8 @@ func autoConvert_v1beta1_AdmissionCheckState_To_v1beta2_AdmissionCheckState(in *
 	out.State = v1beta2.CheckState(in.State)
 	out.LastTransitionTime = in.LastTransitionTime
 	out.Message = in.Message
+	out.RequeueAfterSeconds = (*int32)(unsafe.Pointer(in.RequeueAfterSeconds))
+	out.RetryCount = (*int32)(unsafe.Pointer(in.RetryCount))
 	out.PodSetUpdates = *(*[]v1beta2.PodSetUpdate)(unsafe.Pointer(&in.PodSetUpdates))
 	return nil
 }
@@ -958,6 +960,8 @@ func autoConvert_v1beta2_AdmissionCheckState_To_v1beta1_AdmissionCheckState(in *
 	out.State = CheckState(in.State)
 	out.LastTransitionTime = in.LastTransitionTime
 	out.Message = in.Message
+	out.RequeueAfterSeconds = (*int32)(unsafe.Pointer(in.RequeueAfterSeconds))
+	out.RetryCount = (*int32)(unsafe.Pointer(in.RetryCount))
 	out.PodSetUpdates = *(*[]PodSetUpdate)(unsafe.Pointer(&in.PodSetUpdates))
 	return nil
 }
diff --git a/apis/kueue/v1beta1/zz_generated.deepcopy.go b/apis/kueue/v1beta1/zz_generated.deepcopy.go
index 9961a7c8f..6f98100a4 100644
--- a/apis/kueue/v1beta1/zz_generated.deepcopy.go
+++ b/apis/kueue/v1beta1/zz_generated.deepcopy.go
@@ -151,6 +151,16 @@ func (in *AdmissionCheckSpec) DeepCopy() *AdmissionCheckSpec {
 func (in *AdmissionCheckState) DeepCopyInto(out *AdmissionCheckState) {
 	*out = *in
 	in.LastTransitionTime.DeepCopyInto(&out.LastTransitionTime)
+	if in.RequeueAfterSeconds != nil {
+		in, out := &in.RequeueAfterSeconds, &out.RequeueAfterSeconds
+		*out = new(int32)
+		**out = **in
+	}
+	if in.RetryCount != nil {
+		in, out := &in.RetryCount, &out.RetryCount
+		*out = new(int32)
+		**out = **in
+	}
 	if in.PodSetUpdates != nil {
 		in, out := &in.PodSetUpdates, &out.PodSetUpdates
 		*out = make([]PodSetUpdate, len(*in))
diff --git a/apis/kueue/v1beta2/workload_types.go b/apis/kueue/v1beta2/workload_types.go
index ca9ad1023..f64346144 100644
--- a/apis/kueue/v1beta2/workload_types.go
+++ b/apis/kueue/v1beta2/workload_types.go
@@ -757,6 +757,24 @@ type AdmissionCheckState struct {
 	// +required
 	// +kubebuilder:validation:MaxLength=32768
 	Message string `json:"message" protobuf:"bytes,6,opt,name=message"`
+	// requeueAfterSeconds indicates how long to wait at least before
+	// retrying to admit the workload.
+	// The admission check controllers can set this field when State=Retry
+	// to implement delays between retry attempts.
+	//
+	// If nil when State=Retry, Kueue will retry immediately.
+	// If set, Kueue will add the workload back to the queue after
+	//   lastTransitionTime + RequeueAfterSeconds is over.
+	//
+	// +optional
+	// +kubebuilder:validation:Minimum=0
+	RequeueAfterSeconds *int32 `json:"requeueAfterSeconds,omitempty"`
+	// retryCount tracks retry attempts for this admission check.
+	// Kueue automatically increments the counter whenever the
+	// state transitions to Retry.
+	// +optional
+	// +kubebuilder:validation:Minimum=0
+	RetryCount *int32 `json:"retryCount,omitempty"`
 
 	// podSetUpdates contains a list of pod set modifications suggested by AdmissionChecks.
 	// +optional
diff --git a/apis/kueue/v1beta2/zz_generated.deepcopy.go b/apis/kueue/v1beta2/zz_generated.deepcopy.go
index 8be822c52..6fce0ff12 100644
--- a/apis/kueue/v1beta2/zz_generated.deepcopy.go
+++ b/apis/kueue/v1beta2/zz_generated.deepcopy.go
@@ -146,6 +146,16 @@ func (in *AdmissionCheckSpec) DeepCopy() *AdmissionCheckSpec {
 func (in *AdmissionCheckState) DeepCopyInto(out *AdmissionCheckState) {
 	*out = *in
 	in.LastTransitionTime.DeepCopyInto(&out.LastTransitionTime)
+	if in.RequeueAfterSeconds != nil {
+		in, out := &in.RequeueAfterSeconds, &out.RequeueAfterSeconds
+		*out = new(int32)
+		**out = **in
+	}
+	if in.RetryCount != nil {
+		in, out := &in.RetryCount, &out.RetryCount
+		*out = new(int32)
+		**out = **in
+	}
 	if in.PodSetUpdates != nil {
 		in, out := &in.PodSetUpdates, &out.PodSetUpdates
 		*out = make([]PodSetUpdate, len(*in))
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
index 97ebb2267..504343312 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_workloads.yaml
@@ -8613,6 +8613,27 @@ spec:
                         maxItems: 8
                         type: array
                         x-kubernetes-list-type: atomic
+                      requeueAfterSeconds:
+                        description: |-
+                          requeueAfterSeconds indicates how long to wait at least before
+                          retrying to admit the workload.
+                          The admission check controllers can set this field when State=Retry
+                          to implement delays between retry attempts.
+
+                          If nil when State=Retry, Kueue will retry immediately.
+                          If set, Kueue will add the workload back to the queue after
+                            lastTransitionTime + RequeueAfterSeconds is over.
+                        format: int32
+                        minimum: 0
+                        type: integer
+                      retryCount:
+                        description: |-
+                          retryCount tracks retry attempts for this admission check.
+                          Kueue automatically increments the counter whenever the
+                          state transitions to Retry.
+                        format: int32
+                        minimum: 0
+                        type: integer
                       state:
                         description: state of the admissionCheck, one of Pending, Ready, Retry, Rejected
                         enum:
@@ -17491,6 +17512,27 @@ spec:
                         maxItems: 8
                         type: array
                         x-kubernetes-list-type: atomic
+                      requeueAfterSeconds:
+                        description: |-
+                          requeueAfterSeconds indicates how long to wait at least before
+                          retrying to admit the workload.
+                          The admission check controllers can set this field when State=Retry
+                          to implement delays between retry attempts.
+
+                          If nil when State=Retry, Kueue will retry immediately.
+                          If set, Kueue will add the workload back to the queue after
+                            lastTransitionTime + RequeueAfterSeconds is over.
+                        format: int32
+                        minimum: 0
+                        type: integer
+                      retryCount:
+                        description: |-
+                          retryCount tracks retry attempts for this admission check.
+                          Kueue automatically increments the counter whenever the
+                          state transitions to Retry.
+                        format: int32
+                        minimum: 0
+                        type: integer
                       state:
                         description: state of the admissionCheck, one of Pending, Ready, Retry, Rejected
                         enum:
diff --git a/client-go/applyconfiguration/kueue/v1beta1/admissioncheckstate.go b/client-go/applyconfiguration/kueue/v1beta1/admissioncheckstate.go
index 7a01f579b..f73608e10 100644
--- a/client-go/applyconfiguration/kueue/v1beta1/admissioncheckstate.go
+++ b/client-go/applyconfiguration/kueue/v1beta1/admissioncheckstate.go
@@ -25,11 +25,13 @@ import (
 // AdmissionCheckStateApplyConfiguration represents a declarative configuration of the AdmissionCheckState type for use
 // with apply.
 type AdmissionCheckStateApplyConfiguration struct {
-	Name               *kueuev1beta1.AdmissionCheckReference `json:"name,omitempty"`
-	State              *kueuev1beta1.CheckState              `json:"state,omitempty"`
-	LastTransitionTime *v1.Time                              `json:"lastTransitionTime,omitempty"`
-	Message            *string                               `json:"message,omitempty"`
-	PodSetUpdates      []PodSetUpdateApplyConfiguration      `json:"podSetUpdates,omitempty"`
+	Name                *kueuev1beta1.AdmissionCheckReference `json:"name,omitempty"`
+	State               *kueuev1beta1.CheckState              `json:"state,omitempty"`
+	LastTransitionTime  *v1.Time                              `json:"lastTransitionTime,omitempty"`
+	Message             *string                               `json:"message,omitempty"`
+	RequeueAfterSeconds *int32                                `json:"requeueAfterSeconds,omitempty"`
+	RetryCount          *int32                                `json:"retryCount,omitempty"`
+	PodSetUpdates       []PodSetUpdateApplyConfiguration      `json:"podSetUpdates,omitempty"`
 }
 
 // AdmissionCheckStateApplyConfiguration constructs a declarative configuration of the AdmissionCheckState type for use with
@@ -70,6 +72,22 @@ func (b *AdmissionCheckStateApplyConfiguration) WithMessage(value string) *Admis
 	return b
 }
 
+// WithRequeueAfterSeconds sets the RequeueAfterSeconds field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the RequeueAfterSeconds field is set to the value of the last call.
+func (b *AdmissionCheckStateApplyConfiguration) WithRequeueAfterSeconds(value int32) *AdmissionCheckStateApplyConfiguration {
+	b.RequeueAfterSeconds = &value
+	return b
+}
+
+// WithRetryCount sets the RetryCount field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the RetryCount field is set to the value of the last call.
+func (b *AdmissionCheckStateApplyConfiguration) WithRetryCount(value int32) *AdmissionCheckStateApplyConfiguration {
+	b.RetryCount = &value
+	return b
+}
+
 // WithPodSetUpdates adds the given value to the PodSetUpdates field in the declarative configuration
 // and returns the receiver, so that objects can be build by chaining "With" function invocations.
 // If called multiple times, values provided by each call will be appended to the PodSetUpdates field.
diff --git a/client-go/applyconfiguration/kueue/v1beta2/admissioncheckstate.go b/client-go/applyconfiguration/kueue/v1beta2/admissioncheckstate.go
index b96705847..36afb081e 100644
--- a/client-go/applyconfiguration/kueue/v1beta2/admissioncheckstate.go
+++ b/client-go/applyconfiguration/kueue/v1beta2/admissioncheckstate.go
@@ -25,11 +25,13 @@ import (
 // AdmissionCheckStateApplyConfiguration represents a declarative configuration of the AdmissionCheckState type for use
 // with apply.
 type AdmissionCheckStateApplyConfiguration struct {
-	Name               *kueuev1beta2.AdmissionCheckReference `json:"name,omitempty"`
-	State              *kueuev1beta2.CheckState              `json:"state,omitempty"`
-	LastTransitionTime *v1.Time                              `json:"lastTransitionTime,omitempty"`
-	Message            *string                               `json:"message,omitempty"`
-	PodSetUpdates      []PodSetUpdateApplyConfiguration      `json:"podSetUpdates,omitempty"`
+	Name                *kueuev1beta2.AdmissionCheckReference `json:"name,omitempty"`
+	State               *kueuev1beta2.CheckState              `json:"state,omitempty"`
+	LastTransitionTime  *v1.Time                              `json:"lastTransitionTime,omitempty"`
+	Message             *string                               `json:"message,omitempty"`
+	RequeueAfterSeconds *int32                                `json:"requeueAfterSeconds,omitempty"`
+	RetryCount          *int32                                `json:"retryCount,omitempty"`
+	PodSetUpdates       []PodSetUpdateApplyConfiguration      `json:"podSetUpdates,omitempty"`
 }
 
 // AdmissionCheckStateApplyConfiguration constructs a declarative configuration of the AdmissionCheckState type for use with
@@ -70,6 +72,22 @@ func (b *AdmissionCheckStateApplyConfiguration) WithMessage(value string) *Admis
 	return b
 }
 
+// WithRequeueAfterSeconds sets the RequeueAfterSeconds field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the RequeueAfterSeconds field is set to the value of the last call.
+func (b *AdmissionCheckStateApplyConfiguration) WithRequeueAfterSeconds(value int32) *AdmissionCheckStateApplyConfiguration {
+	b.RequeueAfterSeconds = &value
+	return b
+}
+
+// WithRetryCount sets the RetryCount field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the RetryCount field is set to the value of the last call.
+func (b *AdmissionCheckStateApplyConfiguration) WithRetryCount(value int32) *AdmissionCheckStateApplyConfiguration {
+	b.RetryCount = &value
+	return b
+}
+
 // WithPodSetUpdates adds the given value to the PodSetUpdates field in the declarative configuration
 // and returns the receiver, so that objects can be build by chaining "With" function invocations.
 // If called multiple times, values provided by each call will be appended to the PodSetUpdates field.
diff --git a/cmd/kueueviz/backend/go.mod b/cmd/kueueviz/backend/go.mod
index 643784a23..f2e95f162 100644
--- a/cmd/kueueviz/backend/go.mod
+++ b/cmd/kueueviz/backend/go.mod
@@ -6,9 +6,9 @@ require (
 	github.com/gin-contrib/cors v1.7.6
 	github.com/gin-gonic/gin v1.11.0
 	github.com/gorilla/websocket v1.5.3
-	k8s.io/api v0.32.9
-	k8s.io/apimachinery v0.32.9
-	k8s.io/client-go v0.32.9
+	k8s.io/api v0.32.10
+	k8s.io/apimachinery v0.32.10
+	k8s.io/client-go v0.32.10
 )
 
 require (
diff --git a/cmd/kueueviz/backend/go.sum b/cmd/kueueviz/backend/go.sum
index c37bba1a3..7bc1fe880 100644
--- a/cmd/kueueviz/backend/go.sum
+++ b/cmd/kueueviz/backend/go.sum
@@ -211,12 +211,12 @@ gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.32.9 h1:q/59kk8lnecgG0grJqzrmXC1Jcl2hPWp9ltz0FQuoLI=
-k8s.io/api v0.32.9/go.mod h1:jIfT3rwW4EU1IXZm9qjzSk/2j91k4CJL5vUULrxqp3Y=
-k8s.io/apimachinery v0.32.9 h1:fXk8ktfsxrdThaEOAQFgkhCK7iyoyvS8nbYJ83o/SSs=
-k8s.io/apimachinery v0.32.9/go.mod h1:GpHVgxoKlTxClKcteaeuF1Ul/lDVb74KpZcxcmLDElE=
-k8s.io/client-go v0.32.9 h1:ZMyIQ1TEpTDAQni3L2gH1NZzyOA/gHfNcAazzCxMJ0c=
-k8s.io/client-go v0.32.9/go.mod h1:2OT8aFSYvUjKGadaeT+AVbhkXQSpMAkiSb88Kz2WggI=
+k8s.io/api v0.32.10 h1:ocp4turNfa1V40TuBW/LuA17TeXG9g/GI2ebg0KxBNk=
+k8s.io/api v0.32.10/go.mod h1:AsMsc4b6TuampYqgMEGSv0HBFpRS4BlKTXAVCAa7oF4=
+k8s.io/apimachinery v0.32.10 h1:SAg2kUPLYRcBJQj66oniP1BnXSqw+l1GvJFsJlBmVvQ=
+k8s.io/apimachinery v0.32.10/go.mod h1:GpHVgxoKlTxClKcteaeuF1Ul/lDVb74KpZcxcmLDElE=
+k8s.io/client-go v0.32.10 h1:MFmIjsKtcnn7mStjrJG1ZW2WzLsKKn6ZtL9hHM/W0xU=
+k8s.io/client-go v0.32.10/go.mod h1:qJy/Ws3zSwnu/nD75D+/of1uxbwWHxrYT5P3FuobVLI=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
 k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f h1:GA7//TjRY9yWGy1poLzYYJJ4JRdzg3+O6e8I+e+8T5Y=
diff --git a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
index 2d4a9b04f..93a87ab2f 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_workloads.yaml
@@ -9081,6 +9081,27 @@ spec:
                       maxItems: 8
                       type: array
                       x-kubernetes-list-type: atomic
+                    requeueAfterSeconds:
+                      description: |-
+                        requeueAfterSeconds indicates how long to wait at least before
+                        retrying to admit the workload.
+                        The admission check controllers can set this field when State=Retry
+                        to implement delays between retry attempts.
+
+                        If nil when State=Retry, Kueue will retry immediately.
+                        If set, Kueue will add the workload back to the queue after
+                          lastTransitionTime + RequeueAfterSeconds is over.
+                      format: int32
+                      minimum: 0
+                      type: integer
+                    retryCount:
+                      description: |-
+                        retryCount tracks retry attempts for this admission check.
+                        Kueue automatically increments the counter whenever the
+                        state transitions to Retry.
+                      format: int32
+                      minimum: 0
+                      type: integer
                     state:
                       description: state of the admissionCheck, one of Pending, Ready,
                         Retry, Rejected
@@ -18580,6 +18601,27 @@ spec:
                       maxItems: 8
                       type: array
                       x-kubernetes-list-type: atomic
+                    requeueAfterSeconds:
+                      description: |-
+                        requeueAfterSeconds indicates how long to wait at least before
+                        retrying to admit the workload.
+                        The admission check controllers can set this field when State=Retry
+                        to implement delays between retry attempts.
+
+                        If nil when State=Retry, Kueue will retry immediately.
+                        If set, Kueue will add the workload back to the queue after
+                          lastTransitionTime + RequeueAfterSeconds is over.
+                      format: int32
+                      minimum: 0
+                      type: integer
+                    retryCount:
+                      description: |-
+                        retryCount tracks retry attempts for this admission check.
+                        Kueue automatically increments the counter whenever the
+                        state transitions to Retry.
+                      format: int32
+                      minimum: 0
+                      type: integer
                     state:
                       description: state of the admissionCheck, one of Pending, Ready,
                         Retry, Rejected
diff --git a/keps/2349-multikueue-external-custom-job-support/README.md b/keps/2349-multikueue-external-custom-job-support/README.md
index d9e6aa54a..17ca0e4d6 100644
--- a/keps/2349-multikueue-external-custom-job-support/README.md
+++ b/keps/2349-multikueue-external-custom-job-support/README.md
@@ -158,6 +158,7 @@ Cover the critical part of newly created code.
 - **Feature Gate:** `MultiKueueAdaptersForCustomJobs`
 - Implementation is behind a feature gate.
 - The generic adapter for custom jobs is implemented.
+- *Completed*
 
 #### Beta
 - **Milestone:** v0.15
@@ -177,6 +178,7 @@ Cover the critical part of newly created code.
 
 * 2024-06-20 First draft
 * 2025-07-15 Second draft
+* 2025-11-11 Beta graduation
 
 
 ## Drawbacks
diff --git a/keps/2349-multikueue-external-custom-job-support/kep.yaml b/keps/2349-multikueue-external-custom-job-support/kep.yaml
index 057f15c6a..1ae5d55d5 100644
--- a/keps/2349-multikueue-external-custom-job-support/kep.yaml
+++ b/keps/2349-multikueue-external-custom-job-support/kep.yaml
@@ -16,15 +16,16 @@ see-also:
   - "/keps/693-multikueue"
 
 # The target maturity stage in the current dev cycle for this KEP.
-stage: alpha
+stage: beta
 
 # The most recent milestone for which work toward delivery of this KEP has been
 # done. This can be the current (upcoming) milestone, if it is being actively
 # worked on.
-latest-milestone: "v0.14"
+latest-milestone: "v0.15"
 
 # The milestone at which this feature was, or is targeted to be, at each stage.
 milestone:
   alpha: "v0.14"
+  beta: "v0.15"
 feature-gates:
   - MultiKueueAdaptersForCustomJobs
diff --git a/pkg/controller/admissionchecks/provisioning/controller.go b/pkg/controller/admissionchecks/provisioning/controller.go
index 06064c4ed..0f7514280 100644
--- a/pkg/controller/admissionchecks/provisioning/controller.go
+++ b/pkg/controller/admissionchecks/provisioning/controller.go
@@ -82,8 +82,7 @@ type Controller struct {
 }
 
 type workloadInfo struct {
-	checkStates  []kueue.AdmissionCheckState
-	requeueState *kueue.RequeueState
+	checkStates []kueue.AdmissionCheckState
 }
 
 var _ reconcile.Reconciler = (*Controller)(nil)
@@ -486,7 +485,6 @@ func (wlInfo *workloadInfo) update(wl *kueue.Workload, c clock.Clock) {
 	for _, check := range wl.Status.AdmissionChecks {
 		workload.SetAdmissionCheckState(&wlInfo.checkStates, check, c)
 	}
-	wlInfo.requeueState = wl.Status.RequeueState
 }
 
 func (c *Controller) syncCheckStates(
@@ -499,7 +497,6 @@ func (c *Controller) syncCheckStates(
 	wlInfo.update(wl, c.clock)
 	checksMap := slices.ToRefMap(wl.Status.AdmissionChecks, func(c *kueue.AdmissionCheckState) kueue.AdmissionCheckReference { return c.Name })
 	wlPatch := workload.BaseSSAWorkload(wl, true)
-	wlPatch.Status.RequeueState = wl.Status.RequeueState.DeepCopy()
 	recorderMessages := make([]string, 0, len(checkConfig))
 	updated := false
 	for check, prc := range checkConfig {
@@ -537,11 +534,11 @@ func (c *Controller) syncCheckStates(
 					// it is going to be retried
 					message := fmt.Sprintf("Retrying after failure: %s", apimeta.FindStatusCondition(pr.Status.Conditions, autoscaling.Failed).Message)
 					updated = updateCheckMessage(&checkState, message) || updated
-					if wl.Status.RequeueState == nil || getAttempt(log, pr, wl.Name, check) > ptr.Deref(wl.Status.RequeueState.Count, 0) {
+					if getAttempt(log, pr, wl.Name, check) > ptr.Deref(checkState.RetryCount, 0) {
 						// We don't want to Retry on old ProvisioningRequests
 						updated = true
 						updateCheckState(&checkState, kueue.CheckStateRetry)
-						workload.UpdateRequeueState(wlPatch, backoffBaseSeconds, backoffMaxSeconds, c.clock)
+						workload.UpdateAdmissionCheckRequeueState(&checkState, backoffBaseSeconds, backoffMaxSeconds, c.clock)
 					}
 				} else {
 					updated = true
@@ -562,10 +559,10 @@ func (c *Controller) syncCheckStates(
 						// it is going to be retried
 						message := fmt.Sprintf("Retrying after booking expired: %s", apimeta.FindStatusCondition(pr.Status.Conditions, autoscaling.BookingExpired).Message)
 						updated = updateCheckMessage(&checkState, message) || updated
-						if wl.Status.RequeueState == nil || getAttempt(log, pr, wl.Name, check) > ptr.Deref(wl.Status.RequeueState.Count, 0) {
+						if getAttempt(log, pr, wl.Name, check) > ptr.Deref(checkState.RetryCount, 0) {
 							updated = true
 							updateCheckState(&checkState, kueue.CheckStateRetry)
-							workload.UpdateRequeueState(wlPatch, backoffBaseSeconds, backoffMaxSeconds, c.clock)
+							workload.UpdateAdmissionCheckRequeueState(&checkState, backoffBaseSeconds, backoffMaxSeconds, c.clock)
 						}
 					} else {
 						updated = true
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 378e9cc72..0c01d1186 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -194,6 +194,17 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, nil
 	}
 
+	if requeueAt := workload.NeedsRequeueAtUpdate(&wl, r.clock); requeueAt != nil {
+		err := workload.PatchAdmissionStatus(ctx, r.client, &wl, r.clock, func() (*kueue.Workload, bool, error) {
+			if wl.Status.RequeueState == nil {
+				wl.Status.RequeueState = &kueue.RequeueState{}
+			}
+			log.V(2).Info("At least one admission check set a retry time", "requeueAt", requeueAt, "current", wl.Status.RequeueState.RequeueAt)
+			workload.SetRequeueState(&wl, *requeueAt, false)
+			return &wl, true, nil
+		})
+		return reconcile.Result{}, client.IgnoreNotFound(err)
+	}
 	if features.Enabled(features.DynamicResourceAllocation) && workload.Status(&wl) == workload.StatusPending &&
 		workload.HasDRA(&wl) {
 		workload.AdjustResources(ctx, r.client, &wl)
@@ -264,6 +275,37 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			return ctrl.Result{}, client.IgnoreNotFound(err)
 		}
 
+		if wl.Status.RequeueState != nil && wl.Status.RequeueState.RequeueAt != nil {
+			requeueAfter := ptr.To(wl.Status.RequeueState.RequeueAt.Sub(r.clock.Now()))
+			if requeueAfter != nil && *requeueAfter > 0 {
+				log.V(3).Info("Waiting for backoff to finish", "backoff", *requeueAfter)
+				return reconcile.Result{RequeueAfter: *requeueAfter}, nil
+			}
+
+			if workload.Status(&wl) == workload.StatusPending {
+				log.V(3).Info("Pending workload requeued after backoff")
+
+				// Clear RequeueAt since backoff has elapsed
+				err := workload.PatchAdmissionStatus(ctx, r.client, &wl, r.clock, func() (*kueue.Workload, bool, error) {
+					if wl.Status.RequeueState != nil {
+						wl.Status.RequeueState.RequeueAt = nil
+					}
+					return &wl, true, nil
+				})
+				if err != nil {
+					return ctrl.Result{}, client.IgnoreNotFound(err)
+				}
+
+				if err := r.queues.AddOrUpdateWorkload(wl.DeepCopy()); err != nil {
+					log.V(2).Info("failed to put the workload back into queue", "error", err)
+					return ctrl.Result{}, err
+				}
+
+				log.V(3).Info("Workload requeued after backoff")
+				return ctrl.Result{}, nil
+			}
+		}
+
 		var updated bool
 		var requeueAfter time.Duration
 		err := workload.PatchAdmissionStatus(ctx, r.client, &wl, r.clock, func() (*kueue.Workload, bool, error) {
@@ -315,7 +357,14 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			apimeta.RemoveStatusCondition(&wl.Status.Conditions, kueue.WorkloadDeactivationTarget)
 		}
 		if wl.Status.RequeueState != nil {
-			wl.Status.RequeueState = nil
+			if features.Enabled(features.WorkloadRequestUseMergePatch) {
+				// When using merge patch, we can set the entire RequeueState to nil.
+				wl.Status.RequeueState = nil
+			} else {
+				// When using SSA, we must clear individual fields instead of setting to nil.
+				wl.Status.RequeueState.Count = nil
+				wl.Status.RequeueState.RequeueAt = nil
+			}
 			updated = true
 		}
 		if updated {
@@ -857,6 +906,7 @@ func (r *WorkloadReconciler) Update(e event.TypedUpdateEvent[*kueue.Workload]) b
 			backoff = time.Until(e.ObjectNew.Status.RequeueState.RequeueAt.Time)
 		}
 		immediate := backoff <= 0
+		log.V(3).Info("Workload transitioned back to pending", "backoff", backoff, "immediate", immediate)
 		// trigger the move of associated inadmissibleWorkloads, if there are any.
 		r.queues.QueueAssociatedInadmissibleWorkloadsAfter(ctx, e.ObjectNew, func() {
 			// Delete the workload from cache while holding the queues lock
@@ -865,8 +915,11 @@ func (r *WorkloadReconciler) Update(e event.TypedUpdateEvent[*kueue.Workload]) b
 			if err := r.cache.DeleteWorkload(log, e.ObjectNew); err != nil {
 				log.Error(err, "Failed to delete workload from cache")
 			}
-			// Here we don't take the lock as it is already taken by the wrapping
-			// function.
+			// Here we don't take the lock as it is already taken by the wrapping function.
+			// The delayed requeue is done in the Reconcile function, but we need to
+			// keep the immediate retry here because it will ensure that
+			// AddOrUpdateWorkload is only called once. When moving it to the main
+			// reconciler, we would execute it on every run, which might mess up the state.
 			if immediate {
 				// Skip queue operations for DRA workloads - they are handled in Reconcile loop
 				if features.Enabled(features.DynamicResourceAllocation) && workload.HasDRA(e.ObjectNew) {
@@ -879,26 +932,6 @@ func (r *WorkloadReconciler) Update(e event.TypedUpdateEvent[*kueue.Workload]) b
 				}
 			}
 		})
-
-		if !immediate {
-			log.V(3).Info("Workload to be requeued after backoff", "backoff", backoff, "requeueAt", e.ObjectNew.Status.RequeueState.RequeueAt.Time)
-			// Skip delayed requeue for DRA workloads - they are handled in Reconcile loop
-			if features.Enabled(features.DynamicResourceAllocation) && workload.HasDRA(e.ObjectNew) {
-				log.V(3).Info("Skipping delayed requeue for DRA workload - handled in Reconcile")
-			} else {
-				time.AfterFunc(backoff, func() {
-					updatedWl := kueue.Workload{}
-					err := r.client.Get(ctx, client.ObjectKeyFromObject(e.ObjectNew), &updatedWl)
-					if err == nil && workload.Status(&updatedWl) == workload.StatusPending {
-						if err = r.queues.AddOrUpdateWorkload(wlCopy); err != nil {
-							log.V(2).Info("ignored an error for now", "error", err)
-						} else {
-							log.V(3).Info("Workload requeued after backoff")
-						}
-					}
-				})
-			}
-		}
 	case prevStatus == workload.StatusAdmitted && status == workload.StatusAdmitted && !equality.Semantic.DeepEqual(e.ObjectOld.Status.ReclaimablePods, e.ObjectNew.Status.ReclaimablePods),
 		features.Enabled(features.ElasticJobsViaWorkloadSlices) && workloadslicing.ScaledDown(workload.ExtractPodSetCountsFromWorkload(e.ObjectOld), workload.ExtractPodSetCountsFromWorkload(e.ObjectNew)),
 		workloadPriorityClassChanged(e.ObjectOld, e.ObjectNew):
@@ -969,7 +1002,7 @@ func (r *WorkloadReconciler) SetupWithManager(mgr ctrl.Manager, cfg *config.Conf
 // and calculates the remaining timeout duration.
 //
 // It returns two values:
-//  1. A underlyingCause that complements informarion carried by kueue.WorkloadEvictedByPodsReadyTimeout
+//  1. A underlyingCause that complements information carried by kueue.WorkloadEvictedByPodsReadyTimeout
 //
 // (e.g., WaitForStart, WaitForRecovery, or empty if not applicable).
 //
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index fa6bba45e..26aa0954a 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -314,6 +314,7 @@ var defaultVersionedFeatureGates = map[featuregate.Feature]featuregate.Versioned
 	},
 	MultiKueueAdaptersForCustomJobs: {
 		{Version: version.MustParse("0.14"), Default: false, PreRelease: featuregate.Alpha},
+		{Version: version.MustParse("0.15"), Default: true, PreRelease: featuregate.Beta},
 	},
 	WorkloadRequestUseMergePatch: {
 		{Version: version.MustParse("0.14"), Default: false, PreRelease: featuregate.Alpha},
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 5b8fe45a3..6767fa316 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -877,14 +877,15 @@ const (
 
 func (s *Scheduler) updateEntryPenalty(log logr.Logger, e *entry, op usageOp) {
 	lqKey := utilqueue.NewLocalQueueReference(e.Obj.Namespace, e.Obj.Spec.QueueName)
+	lqObjRef := klog.KRef(e.Obj.Namespace, string(e.Obj.Spec.QueueName))
 	penalty := afs.CalculateEntryPenalty(e.SumTotalRequests(), s.admissionFairSharing)
 
 	switch op {
 	case add:
 		s.queues.PushEntryPenalty(lqKey, penalty)
-		log.V(3).Info("Entry penalty added to lq", "lqKey", lqKey, "penalty", penalty)
+		log.V(3).Info("Entry penalty added to localQueue", "localQueue", lqObjRef, "penalty", penalty)
 	case subtract:
 		s.queues.SubEntryPenalty(lqKey, penalty)
-		log.V(3).Info("Entry penalty subtracted from lq", "lqKey", lqKey, "penalty", penalty)
+		log.V(3).Info("Entry penalty subtracted from localQueue", "localQueue", lqObjRef, "penalty", penalty)
 	}
 }
diff --git a/pkg/workload/admissionchecks.go b/pkg/workload/admissionchecks.go
index 4769739a3..d409e52cc 100644
--- a/pkg/workload/admissionchecks.go
+++ b/pkg/workload/admissionchecks.go
@@ -23,9 +23,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/utils/clock"
+	"k8s.io/utils/ptr"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
 	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
+	"sigs.k8s.io/kueue/pkg/util/wait"
 )
 
 // SyncAdmittedCondition sync the state of the Admitted condition based on the
@@ -90,11 +92,18 @@ func resetChecksOnEviction(w *kueue.Workload, now time.Time) {
 		if checks[i].State == kueue.CheckStatePending {
 			continue
 		}
+		var retryCount *int32
+		if checks[i].State == kueue.CheckStateRetry {
+			tmpRetryCount := ptr.Deref(checks[i].RetryCount, 0) + 1
+			retryCount = ptr.To(tmpRetryCount)
+		}
 		checks[i] = kueue.AdmissionCheckState{
-			Name:               checks[i].Name,
-			State:              kueue.CheckStatePending,
-			LastTransitionTime: metav1.NewTime(now),
-			Message:            "Reset to Pending after eviction. Previously: " + string(checks[i].State),
+			Name:                checks[i].Name,
+			State:               kueue.CheckStatePending,
+			Message:             "Reset to Pending after eviction. Previously: " + string(checks[i].State),
+			LastTransitionTime:  metav1.NewTime(now),
+			RequeueAfterSeconds: checks[i].RequeueAfterSeconds,
+			RetryCount:          retryCount,
 		}
 	}
 }
@@ -123,6 +132,7 @@ func SetAdmissionCheckState(checks *[]kueue.AdmissionCheckState, newCheck kueue.
 	}
 	existingCondition.Message = newCheck.Message
 	existingCondition.PodSetUpdates = newCheck.PodSetUpdates
+	existingCondition.RequeueAfterSeconds = newCheck.RequeueAfterSeconds
 	return true
 }
 
@@ -186,3 +196,66 @@ func HasRejectedChecks(wl *kueue.Workload) bool {
 	}
 	return false
 }
+
+// GetMaxRetryTime returns the max retry time from all the admission checks.
+// It will return a zero time if no retry time is found.
+func GetMaxRetryTime(wl *kueue.Workload) metav1.Time {
+	var max time.Time
+	for i := range wl.Status.AdmissionChecks {
+		if wl.Status.AdmissionChecks[i].RequeueAfterSeconds == nil {
+			continue
+		}
+		// This should never happen, but prevents panics
+		if wl.Status.AdmissionChecks[i].LastTransitionTime.IsZero() {
+			continue
+		}
+		if wl.Status.AdmissionChecks[i].State != kueue.CheckStateRetry {
+			continue
+		}
+
+		retryTime := wl.Status.AdmissionChecks[i].LastTransitionTime.Add(time.Duration(*wl.Status.AdmissionChecks[i].RequeueAfterSeconds) * time.Second)
+		if max.Before(retryTime) {
+			max = retryTime
+		}
+	}
+	return metav1.NewTime(max)
+}
+
+// NeedsRequeueAtUpdate checks if the workload needs its RequeueAt time updated
+// based on admission check retry times. It returns the target requeue time if an
+// update should be performed, or nil if no update is needed.
+func NeedsRequeueAtUpdate(wl *kueue.Workload, clock clock.Clock) *metav1.Time {
+	maxTime := GetMaxRetryTime(wl)
+	// No retry time set
+	if maxTime.IsZero() {
+		return nil
+	}
+	// Retry time is in the past
+	if !maxTime.After(clock.Now()) {
+		return nil
+	}
+	// Check if we need to update RequeueState
+	if wl.Status.RequeueState != nil {
+		currentRequeueAt := ptr.Deref(wl.Status.RequeueState.RequeueAt, metav1.NewTime(time.Time{}))
+		if !currentRequeueAt.Before(&maxTime) || currentRequeueAt.Equal(&maxTime) {
+			// Current time is already >= maxTime, no update needed
+			return nil
+		}
+	}
+	return &maxTime
+}
+
+// UpdateAdmissionCheckRequeueState calculates the RequeueAfterSeconds based on the backoff and requeuingCount
+func UpdateAdmissionCheckRequeueState(acState *kueue.AdmissionCheckState, backoffBaseSeconds int32, backoffMaxSeconds int32, clock clock.Clock) {
+	requeuingCount := ptr.Deref(acState.RetryCount, 0) + 1
+
+	// Every backoff duration is about "60s*2^(n-1)+Rand" where:
+	// - "n" represents the "requeuingCount",
+	// - "Rand" represents the random jitter.
+	// During this time, the workload is treated as inadmissible and other
+	// workloads will have a chance to be admitted.
+	backoff := wait.NewBackoff(time.Duration(backoffBaseSeconds)*time.Second, time.Duration(backoffMaxSeconds)*time.Second, 2, 0.0001)
+	waitDuration := backoff.WaitTime(int(requeuingCount))
+
+	acState.RequeueAfterSeconds = ptr.To(int32(waitDuration.Truncate(time.Second).Seconds()))
+}
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index c2e95af32..9ae185779 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -624,8 +624,31 @@ func UpdateRequeueState(wl *kueue.Workload, backoffBaseSeconds int32, backoffMax
 	backoff := wait.NewBackoff(time.Duration(backoffBaseSeconds)*time.Second, time.Duration(backoffMaxSeconds)*time.Second, 2, 0.0001)
 	waitDuration := backoff.WaitTime(int(requeuingCount))
 
-	wl.Status.RequeueState.RequeueAt = ptr.To(metav1.NewTime(clock.Now().Add(waitDuration)))
-	wl.Status.RequeueState.Count = &requeuingCount
+	_ = SetRequeueState(wl, metav1.NewTime(clock.Now().Add(waitDuration)), true)
+}
+
+// SetRequeueState sets the status.requeueState field with the given timeout
+// if it's greater than the existing value.
+// It will return true if the workload was mutated.
+func SetRequeueState(wl *kueue.Workload, waitUntil metav1.Time, incrementCount bool) bool {
+	if wl.Status.RequeueState == nil {
+		wl.Status.RequeueState = &kueue.RequeueState{}
+	}
+
+	// The requeue state is shared between multiple components,
+	// so we have to ensure that we don't overwrite a future requeue.
+	var updated bool
+	currentRequeueAt := ptr.Deref(wl.Status.RequeueState.RequeueAt, metav1.NewTime(time.Time{}))
+	if currentRequeueAt.Before(&waitUntil) && !currentRequeueAt.Equal(&waitUntil) {
+		wl.Status.RequeueState.RequeueAt = &waitUntil
+		updated = true
+	}
+	if incrementCount {
+		requeuingCount := ptr.Deref(wl.Status.RequeueState.Count, 0) + 1
+		wl.Status.RequeueState.Count = &requeuingCount
+		updated = true
+	}
+	return updated
 }
 
 // SetRequeuedCondition sets the WorkloadRequeued condition to true
@@ -863,7 +886,10 @@ func PropagateResourceRequests(w *kueue.Workload, info *Info) bool {
 // If strict is true, resourceVersion will be part of the patch.
 func admissionStatusPatch(w *kueue.Workload, wlCopy *kueue.Workload) {
 	wlCopy.Status.Admission = w.Status.Admission.DeepCopy()
-	wlCopy.Status.RequeueState = w.Status.RequeueState.DeepCopy()
+	// Only include RequeueState in the patch if it has meaningful content.
+	if w.Status.RequeueState != nil && (w.Status.RequeueState.Count != nil || w.Status.RequeueState.RequeueAt != nil) {
+		wlCopy.Status.RequeueState = w.Status.RequeueState.DeepCopy()
+	}
 	if wlCopy.Status.Admission != nil {
 		// Clear ResourceRequests; Assignment.PodSetAssignment[].ResourceUsage supercedes it
 		wlCopy.Status.ResourceRequests = []kueue.PodSetRequest{}
@@ -1444,3 +1470,24 @@ func ReasonWithCause(reason, underlyingCause string) string {
 func ClusterName(wl *kueue.Workload) string {
 	return ptr.Deref(wl.Status.ClusterName, "")
 }
+
+// ResetRequeue resets the requeue state of the workload as well as all admission checks
+// It returns true if the workload was modified.
+func ResetRequeue(wl *kueue.Workload) bool {
+	var updated bool
+
+	if wl.Status.RequeueState != nil {
+		wl.Status.RequeueState = nil
+		updated = true
+	}
+	for i := range wl.Status.AdmissionChecks {
+		if wl.Status.AdmissionChecks[i].RequeueAfterSeconds == nil || wl.Status.AdmissionChecks[i].RetryCount == nil {
+			continue
+		}
+		wl.Status.AdmissionChecks[i].RequeueAfterSeconds = nil
+		wl.Status.AdmissionChecks[i].RetryCount = nil
+		updated = true
+	}
+
+	return updated
+}
diff --git a/site/content/en/docs/reference/kueue-config.v1beta2.md b/site/content/en/docs/reference/kueue-config.v1beta2.md
index 667301d81..40391b9ad 100644
--- a/site/content/en/docs/reference/kueue-config.v1beta2.md
+++ b/site/content/en/docs/reference/kueue-config.v1beta2.md
@@ -842,10 +842,9 @@ Defaults to 5min.</p>
 <code>bool</code>
 </td>
 <td>
-   <p>BlockAdmission when true, cluster queue will block admissions for all
+   <p>BlockAdmission when true, the cluster queue will block admissions for all
 subsequent jobs until the jobs reach the PodsReady=true condition.
-This setting is only honored when <code>Enable</code> is set to true.
-Defaults to true.</p>
+Defaults to false.</p>
 </td>
 </tr>
 <tr><td><code>requeuingStrategy</code><br/>
@@ -864,8 +863,7 @@ last transition to the PodsReady=false condition after a Workload is Admitted an
 Such a transition may happen when a Pod failed and the replacement Pod
 is awaited to be scheduled.
 After exceeding the timeout the corresponding job gets suspended again
-and requeued after the backoff delay. The timeout is enforced only if waitForPodsReady.enable=true.
-If not set, there is no timeout.</p>
+and requeued after the backoff delay.</p>
 </td>
 </tr>
 </tbody>
diff --git a/site/content/en/docs/reference/kueue.v1beta1.md b/site/content/en/docs/reference/kueue.v1beta1.md
index d523eb9aa..7f56da799 100644
--- a/site/content/en/docs/reference/kueue.v1beta1.md
+++ b/site/content/en/docs/reference/kueue.v1beta1.md
@@ -562,6 +562,28 @@ This should be when the underlying condition changed.  If that is not known, the
 This may be an empty string.</p>
 </td>
 </tr>
+<tr><td><code>requeueAfterSeconds</code><br/>
+<code>int32</code>
+</td>
+<td>
+   <p>requeueAfterSeconds indicates how long to wait at least before
+retrying to admit the workload.
+The admission check controllers can set this field when State=Retry
+to implement delays between retry attempts.</p>
+<p>If nil when State=Retry, Kueue will retry immediately.
+If set, Kueue will add the workload back to the queue after
+lastTransitionTime + RequeueAfterSeconds is over.</p>
+</td>
+</tr>
+<tr><td><code>retryCount</code><br/>
+<code>int32</code>
+</td>
+<td>
+   <p>retryCount tracks retry attempts for this admission check.
+Kueue automatically increments the counter whenever the
+state transitions to Retry.</p>
+</td>
+</tr>
 <tr><td><code>podSetUpdates</code><br/>
 <a href="#kueue-x-k8s-io-v1beta1-PodSetUpdate"><code>[]PodSetUpdate</code></a>
 </td>
diff --git a/site/content/en/docs/reference/kueue.v1beta2.md b/site/content/en/docs/reference/kueue.v1beta2.md
index 9d1e31048..11c7fdd69 100644
--- a/site/content/en/docs/reference/kueue.v1beta2.md
+++ b/site/content/en/docs/reference/kueue.v1beta2.md
@@ -551,6 +551,28 @@ This should be when the underlying condition changed.  If that is not known, the
 This may be an empty string.</p>
 </td>
 </tr>
+<tr><td><code>requeueAfterSeconds</code><br/>
+<code>int32</code>
+</td>
+<td>
+   <p>requeueAfterSeconds indicates how long to wait at least before
+retrying to admit the workload.
+The admission check controllers can set this field when State=Retry
+to implement delays between retry attempts.</p>
+<p>If nil when State=Retry, Kueue will retry immediately.
+If set, Kueue will add the workload back to the queue after
+lastTransitionTime + RequeueAfterSeconds is over.</p>
+</td>
+</tr>
+<tr><td><code>retryCount</code><br/>
+<code>int32</code>
+</td>
+<td>
+   <p>retryCount tracks retry attempts for this admission check.
+Kueue automatically increments the counter whenever the
+state transitions to Retry.</p>
+</td>
+</tr>
 <tr><td><code>podSetUpdates</code><br/>
 <a href="#kueue-x-k8s-io-v1beta2-PodSetUpdate"><code>[]PodSetUpdate</code></a>
 </td>
diff --git a/site/content/en/docs/tasks/manage/administer_cluster_quotas.md b/site/content/en/docs/tasks/manage/administer_cluster_quotas.md
index 28f6521ee..0a89fb54e 100644
--- a/site/content/en/docs/tasks/manage/administer_cluster_quotas.md
+++ b/site/content/en/docs/tasks/manage/administer_cluster_quotas.md
@@ -29,7 +29,7 @@ You can perform all these steps at once by
 applying [examples/admin/single-clusterqueue-setup.yaml](/examples/admin/single-clusterqueue-setup.yaml):
 
 ```shell
-kubectl apply -f examples/admin/single-clusterqueue-setup.yaml
+kubectl apply -f https://kueue.sigs.k8s.io/examples/admin/single-clusterqueue-setup.yaml
 ```
 
 ### 1. Create a [ClusterQueue](/docs/concepts/cluster_queue)
diff --git a/site/content/en/docs/tasks/run/deployment.md b/site/content/en/docs/tasks/run/deployment.md
index 5b5310fb2..b1a589693 100644
--- a/site/content/en/docs/tasks/run/deployment.md
+++ b/site/content/en/docs/tasks/run/deployment.md
@@ -88,5 +88,5 @@ Here is a sample Deployment:
 
 You can create the Deployment using the following command:
 ```sh
-kubectl create -f sample-deployment.yaml
+kubectl create -f https://kueue.sigs.k8s.io/examples/serving-workloads/sample-deployment.yaml
 ```
diff --git a/site/content/en/docs/tasks/run/plain_pods.md b/site/content/en/docs/tasks/run/plain_pods.md
index d748ec5b3..2bf083863 100644
--- a/site/content/en/docs/tasks/run/plain_pods.md
+++ b/site/content/en/docs/tasks/run/plain_pods.md
@@ -118,7 +118,7 @@ Here is a sample Pod that just sleeps for a few seconds:
 You can create the Pod using the following command:
 ```sh
 # Create the pod
-kubectl create -f kueue-pod.yaml
+kubectl create -f https://kueue.sigs.k8s.io/examples/pods-kueue/kueue-pod.yaml
 ```
 
 ## Running a group of Pods to be admitted together
@@ -177,7 +177,7 @@ Here is a sample Pod group that just sleeps for a few seconds:
 
 You can create the Pod group using the following command:
 ```sh
-kubectl create -f kueue-pod-group.yaml
+kubectl create -f https://kueue.sigs.k8s.io/examples/pods-kueue/kueue-pod-group.yaml
 ```
 
 The name of the associated Workload created by Kueue equals the name of the Pod



Test patch:
diff --git a/apis/config/v1beta1/configuration_conversion_test.go b/apis/config/v1beta1/configuration_conversion_test.go
index 61a199299..47da3a65c 100644
--- a/apis/config/v1beta1/configuration_conversion_test.go
+++ b/apis/config/v1beta1/configuration_conversion_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"k8s.io/utils/ptr"
 
 	"sigs.k8s.io/kueue/apis/config/v1beta2"
 )
@@ -64,7 +65,9 @@ func TestConfigurationQueueConvertTo(t *testing.T) {
 				},
 			},
 			expected: &v1beta2.Configuration{
-				WaitForPodsReady: &v1beta2.WaitForPodsReady{},
+				WaitForPodsReady: &v1beta2.WaitForPodsReady{
+					BlockAdmission: ptr.To(true),
+				},
 			},
 		},
 		"with WaitForPodsReady disabled": {
@@ -120,7 +123,8 @@ func TestConfigurationQueueConvertFrom(t *testing.T) {
 			},
 			expected: &Configuration{
 				WaitForPodsReady: &WaitForPodsReady{
-					Enable: true,
+					Enable:         true,
+					BlockAdmission: ptr.To(false),
 				},
 			},
 		},
@@ -154,13 +158,6 @@ func TestConfigurationQueueConversion_RoundTrip(t *testing.T) {
 				},
 			},
 		},
-		"with WaitForPodsReady": {
-			v1beta1Obj: &Configuration{
-				WaitForPodsReady: &WaitForPodsReady{
-					Enable: true,
-				},
-			},
-		},
 	}
 
 	for name, tc := range testCases {
diff --git a/apis/config/v1beta2/defaults_test.go b/apis/config/v1beta2/defaults_test.go
index 1083ec1eb..71fd254f9 100644
--- a/apis/config/v1beta2/defaults_test.go
+++ b/apis/config/v1beta2/defaults_test.go
@@ -354,7 +354,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 			},
 			want: &Configuration{
 				WaitForPodsReady: &WaitForPodsReady{
-					BlockAdmission:  ptr.To(true),
+					BlockAdmission:  ptr.To(false),
 					Timeout:         &podsReadyTimeout,
 					RecoveryTimeout: nil,
 					RequeuingStrategy: &RequeuingStrategy{
@@ -391,7 +391,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 			},
 			want: &Configuration{
 				WaitForPodsReady: &WaitForPodsReady{
-					BlockAdmission:  ptr.To(true),
+					BlockAdmission:  ptr.To(false),
 					Timeout:         &podsReadyTimeoutOverwrite,
 					RecoveryTimeout: &metav1.Duration{Duration: time.Minute},
 					RequeuingStrategy: &RequeuingStrategy{
diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go
index 75adf8ffd..499871fcd 100644
--- a/pkg/config/config_test.go
+++ b/pkg/config/config_test.go
@@ -202,7 +202,7 @@ apiVersion: config.kueue.x-k8s.io/v1beta2
 kind: Configuration
 waitForPodsReady:
   timeout: 50s
-  blockAdmission: false
+  blockAdmission: true
   recoveryTimeout: 3m
   requeuingStrategy:
     timestamp: Creation
@@ -619,7 +619,7 @@ objectRetentionPolicies:
 				ManageJobsWithoutQueueName: false,
 				InternalCertManagement:     enableDefaultInternalCertManagement,
 				WaitForPodsReady: &configapi.WaitForPodsReady{
-					BlockAdmission:  ptr.To(false),
+					BlockAdmission:  ptr.To(true),
 					Timeout:         &metav1.Duration{Duration: 50 * time.Second},
 					RecoveryTimeout: &metav1.Duration{Duration: 3 * time.Minute},
 					RequeuingStrategy: &configapi.RequeuingStrategy{
diff --git a/pkg/controller/admissionchecks/provisioning/controller_test.go b/pkg/controller/admissionchecks/provisioning/controller_test.go
index 0ada52b1d..b61c116ac 100644
--- a/pkg/controller/admissionchecks/provisioning/controller_test.go
+++ b/pkg/controller/admissionchecks/provisioning/controller_test.go
@@ -226,9 +226,10 @@ func TestReconcile(t *testing.T) {
 
 	baseConfig := utiltestingapi.MakeProvisioningRequestConfig("config1").ProvisioningClass("class1").WithParameter("p1", "v1")
 
+	var backoffBaseSeconds int32 = 60
 	baseConfigWithRetryStrategy := baseConfig.Clone().RetryStrategy(&kueue.ProvisioningRequestRetryStrategy{
 		BackoffLimitCount:  ptr.To[int32](3),
-		BackoffBaseSeconds: ptr.To[int32](60),
+		BackoffBaseSeconds: ptr.To[int32](backoffBaseSeconds),
 		BackoffMaxSeconds:  ptr.To[int32](1800),
 	})
 
@@ -567,14 +568,14 @@ func TestReconcile(t *testing.T) {
 			wantWorkloads: map[string]*kueue.Workload{
 				baseWorkload.GetName(): (&utiltestingapi.WorkloadWrapper{Workload: *baseWorkload.DeepCopy()}).
 					AdmissionChecks(kueue.AdmissionCheckState{
-						Name:    "check1",
-						State:   kueue.CheckStateRetry,
-						Message: "Retrying after failure: ",
+						Name:                "check1",
+						State:               kueue.CheckStateRetry,
+						Message:             "Retrying after failure: ",
+						RequeueAfterSeconds: ptr.To(backoffBaseSeconds),
 					}, kueue.AdmissionCheckState{
 						Name:  "not-provisioning",
 						State: kueue.CheckStatePending,
 					}).
-					RequeueState(ptr.To[int32](1), nil).
 					Obj(),
 			},
 		},
@@ -999,14 +1000,14 @@ func TestReconcile(t *testing.T) {
 			wantWorkloads: map[string]*kueue.Workload{
 				baseWorkload.GetName(): (&utiltestingapi.WorkloadWrapper{Workload: *baseWorkload.DeepCopy()}).
 					AdmissionChecks(kueue.AdmissionCheckState{
-						Name:    "check1",
-						State:   kueue.CheckStateRetry,
-						Message: "Retrying after booking expired: ",
+						Name:                "check1",
+						State:               kueue.CheckStateRetry,
+						Message:             "Retrying after booking expired: ",
+						RequeueAfterSeconds: ptr.To(backoffBaseSeconds),
 					}, kueue.AdmissionCheckState{
 						Name:  "not-provisioning",
 						State: kueue.CheckStatePending,
 					}).
-					RequeueState(ptr.To[int32](1), nil).
 					Admitted(false).
 					Obj(),
 			},
diff --git a/pkg/controller/core/workload_controller_test.go b/pkg/controller/core/workload_controller_test.go
index 150886f18..a7414fd70 100644
--- a/pkg/controller/core/workload_controller_test.go
+++ b/pkg/controller/core/workload_controller_test.go
@@ -337,10 +337,9 @@ func TestSyncCheckStates(t *testing.T) {
 		"preserve conditions data": {
 			states: []kueue.AdmissionCheckState{
 				{
-					Name:               "ac0",
-					State:              kueue.CheckStateReady,
-					Message:            "Message one",
-					LastTransitionTime: metav1.NewTime(now),
+					Name:    "ac0",
+					State:   kueue.CheckStateReady,
+					Message: "Message one",
 				},
 				{
 					Name:  "ac1",
@@ -351,10 +350,9 @@ func TestSyncCheckStates(t *testing.T) {
 			wantChange: false,
 			wantStates: []kueue.AdmissionCheckState{
 				{
-					Name:               "ac0",
-					State:              kueue.CheckStateReady,
-					Message:            "Message one",
-					LastTransitionTime: metav1.NewTime(now),
+					Name:    "ac0",
+					State:   kueue.CheckStateReady,
+					Message: "Message one",
 				},
 				{
 					Name:  "ac1",
@@ -880,9 +878,10 @@ func TestReconcile(t *testing.T) {
 						Message: "Reset to Pending after eviction. Previously: Rejected",
 					},
 					kueue.AdmissionCheckState{
-						Name:    "check-2",
-						State:   kueue.CheckStatePending,
-						Message: "Reset to Pending after eviction. Previously: Retry",
+						Name:       "check-2",
+						State:      kueue.CheckStatePending,
+						Message:    "Reset to Pending after eviction. Previously: Retry",
+						RetryCount: ptr.To(int32(1)),
 					},
 				).
 				Conditions(
@@ -920,9 +919,10 @@ func TestReconcile(t *testing.T) {
 						Message: "Reset to Pending after eviction. Previously: Rejected",
 					},
 					kueue.AdmissionCheckState{
-						Name:    "check-2",
-						State:   kueue.CheckStatePending,
-						Message: "Reset to Pending after eviction. Previously: Retry",
+						Name:       "check-2",
+						State:      kueue.CheckStatePending,
+						Message:    "Reset to Pending after eviction. Previously: Retry",
+						RetryCount: ptr.To(int32(1)),
 					},
 				).
 				Conditions(
@@ -966,9 +966,10 @@ func TestReconcile(t *testing.T) {
 				ReserveQuota(utiltestingapi.MakeAdmission("q1").Obj()).
 				ControllerReference(batchv1.SchemeGroupVersion.WithKind("Job"), "ownername", "owneruid").
 				AdmissionChecks(kueue.AdmissionCheckState{
-					Name:    "check-1",
-					State:   kueue.CheckStatePending,
-					Message: "Reset to Pending after eviction. Previously: Retry",
+					Name:       "check-1",
+					State:      kueue.CheckStatePending,
+					Message:    "Reset to Pending after eviction. Previously: Retry",
+					RetryCount: ptr.To(int32(1)),
 				}, kueue.AdmissionCheckState{
 					Name:    "check-2",
 					State:   kueue.CheckStatePending,
@@ -1102,7 +1103,7 @@ func TestReconcile(t *testing.T) {
 					Message:            "Admitted by ClusterQueue q1",
 				}).
 				Admitted(true).
-				RequeueState(ptr.To[int32](1), ptr.To(metav1.NewTime(testStartTime.Add(1*time.Second).Truncate(time.Second)))).
+				RequeueState(ptr.To[int32](1), ptr.To(metav1.NewTime(testStartTime.Add(-1*time.Second).Truncate(time.Second)))).
 				Obj(),
 			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
 				ReserveQuota(utiltestingapi.MakeAdmission("q1").Obj()).
@@ -1145,7 +1146,7 @@ func TestReconcile(t *testing.T) {
 					Message:            "Admitted by ClusterQueue q1",
 				}).
 				Admitted(true).
-				RequeueState(ptr.To[int32](10), ptr.To(metav1.NewTime(testStartTime.Add(1*time.Second).Truncate(time.Second)))).
+				RequeueState(ptr.To[int32](10), ptr.To(metav1.NewTime(testStartTime.Add(-1*time.Second).Truncate(time.Second)))).
 				Obj(),
 			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
 				ReserveQuota(utiltestingapi.MakeAdmission("q1").Obj()).
@@ -1310,7 +1311,6 @@ func TestReconcile(t *testing.T) {
 					Reason:  kueue.WorkloadEvictedByPodsReadyTimeout,
 					Message: "Exceeded the PodsReady timeout ns",
 				}).
-				RequeueState(ptr.To[int32](1), ptr.To(metav1.NewTime(testStartTime.Truncate(time.Second)))).
 				Obj(),
 			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
 				Active(true).
@@ -1320,7 +1320,6 @@ func TestReconcile(t *testing.T) {
 					Reason:  kueue.WorkloadBackoffFinished,
 					Message: "The workload backoff was finished",
 				}).
-				RequeueState(ptr.To[int32](1), ptr.To(metav1.NewTime(testStartTime.Truncate(time.Second)))).
 				Obj(),
 		},
 		"should keep the WorkloadRequeued condition until the AdmissionCheck backoff expires": {
@@ -1355,7 +1354,6 @@ func TestReconcile(t *testing.T) {
 					Reason:  kueue.WorkloadEvictedByAdmissionCheck,
 					Message: "Exceeded the AdmissionCheck timeout ns",
 				}).
-				RequeueState(ptr.To[int32](1), ptr.To(metav1.NewTime(testStartTime.Truncate(time.Second)))).
 				Obj(),
 			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
 				Active(true).
@@ -1365,7 +1363,6 @@ func TestReconcile(t *testing.T) {
 					Reason:  kueue.WorkloadBackoffFinished,
 					Message: "The workload backoff was finished",
 				}).
-				RequeueState(ptr.To[int32](1), ptr.To(metav1.NewTime(testStartTime.Truncate(time.Second)))).
 				Obj(),
 		},
 		"shouldn't set the WorkloadRequeued condition when backoff expires and workload finished": {
@@ -2414,6 +2411,160 @@ func TestReconcile(t *testing.T) {
 			},
 			wantError: nil,
 		},
+		"should update requeueState.requeueAt when admission check sets a delay": {
+			workload: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(5*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(10)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(10*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(10)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkloadUseMergePatch: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(10*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(10)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantResult: reconcile.Result{},
+		},
+		"should not update requeueState.requeueAt when admission check delay is smaller": {
+			workload: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(10*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(10*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkloadUseMergePatch: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(10*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantResult: reconcile.Result{RequeueAfter: 10 * time.Second},
+		},
+		"should always use the biggest admission check delay": {
+			workload: utiltestingapi.MakeWorkload("wl", "ns").
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(7)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(10*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(7)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkloadUseMergePatch: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(10*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(7)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantResult: reconcile.Result{},
+		},
+		"should use the biggest total time not the biggest RequeueAfterSeconds": {
+			workload: utiltestingapi.MakeWorkload("wl", "ns").
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime.Add(10 * time.Second)),
+				}).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(10)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkload: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(15*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime.Add(10 * time.Second)),
+				}).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(10)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantWorkloadUseMergePatch: utiltestingapi.MakeWorkload("wl", "ns").
+				RequeueState(nil, ptr.To(metav1.NewTime(testStartTime.Add(15*time.Second)))).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(5)),
+					LastTransitionTime:  metav1.NewTime(testStartTime.Add(10 * time.Second)),
+				}).
+				AdmissionCheck(kueue.AdmissionCheckState{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					RequeueAfterSeconds: ptr.To(int32(10)),
+					LastTransitionTime:  metav1.NewTime(testStartTime),
+				}).
+				Obj(),
+			wantResult: reconcile.Result{},
+		},
 	}
 	for name, tc := range cases {
 		for _, enabled := range []bool{false, true} {
diff --git a/pkg/workload/admissionchecks_test.go b/pkg/workload/admissionchecks_test.go
index 32256438a..20abd54a9 100644
--- a/pkg/workload/admissionchecks_test.go
+++ b/pkg/workload/admissionchecks_test.go
@@ -552,3 +552,127 @@ func TestSetCheckState(t *testing.T) {
 		})
 	}
 }
+
+func TestGetMaxRetryTime(t *testing.T) {
+	baseTime := time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC)
+
+	cases := map[string]struct {
+		checkStates   []kueue.AdmissionCheckState
+		wantRetryTime metav1.Time
+	}{
+		"no admission checks": {
+			checkStates:   []kueue.AdmissionCheckState{},
+			wantRetryTime: metav1.NewTime(time.Time{}),
+		},
+		"admission check with nil RequeueAfterSeconds": {
+			checkStates: []kueue.AdmissionCheckState{
+				{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: nil,
+				},
+			},
+			wantRetryTime: metav1.NewTime(time.Time{}),
+		},
+		"admission check with zero LastTransitionTime": {
+			checkStates: []kueue.AdmissionCheckState{
+				{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(time.Time{}),
+					RequeueAfterSeconds: ptr.To[int32](60),
+				},
+			},
+			wantRetryTime: metav1.NewTime(time.Time{}),
+		},
+		"single admission check with valid retry time": {
+			checkStates: []kueue.AdmissionCheckState{
+				{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: ptr.To[int32](60),
+				},
+			},
+			wantRetryTime: metav1.NewTime(baseTime.Add(60 * time.Second)),
+		},
+		"multiple admission checks, returns max retry time": {
+			checkStates: []kueue.AdmissionCheckState{
+				{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: ptr.To[int32](60),
+				},
+				{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: ptr.To[int32](120),
+				},
+				{
+					Name:                "check3",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: ptr.To[int32](30),
+				},
+			},
+			wantRetryTime: metav1.NewTime(baseTime.Add(120 * time.Second)),
+		},
+		"multiple checks with mixed nil and valid values": {
+			checkStates: []kueue.AdmissionCheckState{
+				{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: nil,
+				},
+				{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: ptr.To[int32](90),
+				},
+				{
+					Name:                "check3",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(time.Time{}),
+					RequeueAfterSeconds: ptr.To[int32](180),
+				},
+			},
+			wantRetryTime: metav1.NewTime(baseTime.Add(90 * time.Second)),
+		},
+		"checks with different base times": {
+			checkStates: []kueue.AdmissionCheckState{
+				{
+					Name:                "check1",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime),
+					RequeueAfterSeconds: ptr.To[int32](60),
+				},
+				{
+					Name:                "check2",
+					State:               kueue.CheckStateRetry,
+					LastTransitionTime:  metav1.NewTime(baseTime.Add(30 * time.Second)),
+					RequeueAfterSeconds: ptr.To[int32](60),
+				},
+			},
+			wantRetryTime: metav1.NewTime(baseTime.Add(90 * time.Second)),
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			wl := utiltestingapi.MakeWorkload("foo", "bar").
+				AdmissionChecks(tc.checkStates...).
+				Obj()
+
+			gotRetryTime := GetMaxRetryTime(wl)
+
+			if !gotRetryTime.Equal(&tc.wantRetryTime) {
+				t.Errorf("GetMaxRetryTime() = %v, want %v", gotRetryTime, tc.wantRetryTime)
+			}
+		})
+	}
+}
diff --git a/pkg/workload/workload_test.go b/pkg/workload/workload_test.go
index c55379139..db014d45a 100644
--- a/pkg/workload/workload_test.go
+++ b/pkg/workload/workload_test.go
@@ -1611,6 +1611,7 @@ func TestSetQuotaReservation(t *testing.T) {
 		})
 	}
 }
+
 func TestPatchAdmissionStatus(t *testing.T) {
 	now := time.Now()
 	fakeClock := testingclock.NewFakeClock(now)
@@ -1674,3 +1675,156 @@ func TestPatchAdmissionStatus(t *testing.T) {
 		}
 	}
 }
+
+func TestSetRequeueState(t *testing.T) {
+	baseTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
+	futureTime := baseTime.Add(5 * time.Minute)
+	evenMoreFutureTime := baseTime.Add(10 * time.Minute)
+	pastTime := baseTime.Add(-5 * time.Minute)
+
+	cases := map[string]struct {
+		workload       *kueue.Workload
+		waitUntil      metav1.Time
+		incrementCount bool
+		wantUpdated    bool
+		wantRequeueAt  *metav1.Time
+		wantCount      *int32
+	}{
+		"should initialize and set time when requeue state is nil without increment": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: nil,
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: false,
+			wantUpdated:    true,
+			wantRequeueAt:  ptr.To(metav1.NewTime(futureTime)),
+			wantCount:      nil,
+		},
+		"should initialize and set time and count when requeue state is nil with increment": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: nil,
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: true,
+			wantUpdated:    true,
+			wantRequeueAt:  ptr.To(metav1.NewTime(futureTime)),
+			wantCount:      ptr.To[int32](1),
+		},
+		"should update time when existing requeue time is earlier": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: &kueue.RequeueState{
+						RequeueAt: ptr.To(metav1.NewTime(pastTime)),
+						Count:     ptr.To[int32](2),
+					},
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: false,
+			wantUpdated:    true,
+			wantRequeueAt:  ptr.To(metav1.NewTime(futureTime)),
+			wantCount:      ptr.To[int32](2),
+		},
+		"should not update time when existing requeue time is later": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: &kueue.RequeueState{
+						RequeueAt: ptr.To(metav1.NewTime(evenMoreFutureTime)),
+						Count:     ptr.To[int32](3),
+					},
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: false,
+			wantUpdated:    false,
+			wantRequeueAt:  ptr.To(metav1.NewTime(evenMoreFutureTime)),
+			wantCount:      ptr.To[int32](3),
+		},
+		"should increment count but keep later requeue time": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: &kueue.RequeueState{
+						RequeueAt: ptr.To(metav1.NewTime(evenMoreFutureTime)),
+						Count:     ptr.To[int32](3),
+					},
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: true,
+			wantUpdated:    true,
+			wantRequeueAt:  ptr.To(metav1.NewTime(evenMoreFutureTime)),
+			wantCount:      ptr.To[int32](4),
+		},
+		"should increment count when requeue time is same": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: &kueue.RequeueState{
+						RequeueAt: ptr.To(metav1.NewTime(futureTime)),
+						Count:     ptr.To[int32](1),
+					},
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: true,
+			wantUpdated:    true,
+			wantRequeueAt:  ptr.To(metav1.NewTime(futureTime)),
+			wantCount:      ptr.To[int32](2),
+		},
+		"should increment from zero count": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: &kueue.RequeueState{
+						RequeueAt: ptr.To(metav1.NewTime(pastTime)),
+						Count:     ptr.To[int32](0),
+					},
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: true,
+			wantUpdated:    true,
+			wantRequeueAt:  ptr.To(metav1.NewTime(futureTime)),
+			wantCount:      ptr.To[int32](1),
+		},
+		"should handle zero time in requeue state": {
+			workload: &kueue.Workload{
+				Status: kueue.WorkloadStatus{
+					RequeueState: &kueue.RequeueState{
+						RequeueAt: ptr.To(metav1.NewTime(time.Time{})),
+						Count:     ptr.To[int32](0),
+					},
+				},
+			},
+			waitUntil:      metav1.NewTime(futureTime),
+			incrementCount: true,
+			wantUpdated:    true,
+			wantRequeueAt:  ptr.To(metav1.NewTime(futureTime)),
+			wantCount:      ptr.To[int32](1),
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			gotUpdated := SetRequeueState(tc.workload, tc.waitUntil, tc.incrementCount)
+
+			if gotUpdated != tc.wantUpdated {
+				t.Errorf("SetRequeueState() returned %v, want %v", gotUpdated, tc.wantUpdated)
+			}
+
+			if tc.workload.Status.RequeueState == nil {
+				t.Fatal("RequeueState should not be nil after SetRequeueState")
+			}
+
+			if diff := cmp.Diff(tc.wantRequeueAt, tc.workload.Status.RequeueState.RequeueAt); diff != "" {
+				t.Errorf("Unexpected RequeueAt (-want +got):\n%s", diff)
+			}
+
+			if diff := cmp.Diff(tc.wantCount, tc.workload.Status.RequeueState.Count); diff != "" {
+				t.Errorf("Unexpected Count (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/test/integration/multikueue/external_job_test.go b/test/integration/multikueue/external_job_test.go
index e3903beef..b24cb0bf8 100644
--- a/test/integration/multikueue/external_job_test.go
+++ b/test/integration/multikueue/external_job_test.go
@@ -42,7 +42,6 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	workloadrayjob "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob"
 	dispatcher "sigs.k8s.io/kueue/pkg/controller/workloaddispatcher"
-	"sigs.k8s.io/kueue/pkg/features"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	utiltestingapi "sigs.k8s.io/kueue/pkg/util/testing/v1beta2"
 	testingrayjob "sigs.k8s.io/kueue/pkg/util/testingjobs/rayjob"
@@ -146,7 +145,6 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 		})
 
 		ginkgo.BeforeEach(func() {
-			features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.MultiKueueAdaptersForCustomJobs, true)
 			managerNs = util.CreateNamespaceFromPrefixWithLog(managerTestCluster.ctx, managerTestCluster.client, "multikueue-")
 			worker1Ns = util.CreateNamespaceWithLog(worker1TestCluster.ctx, worker1TestCluster.client, managerNs.Name)
 			worker2Ns = util.CreateNamespaceWithLog(worker2TestCluster.ctx, worker2TestCluster.client, managerNs.Name)
diff --git a/test/integration/multikueue/jobs_test.go b/test/integration/multikueue/jobs_test.go
index 3a6b8fcb3..3f1d31c11 100644
--- a/test/integration/multikueue/jobs_test.go
+++ b/test/integration/multikueue/jobs_test.go
@@ -1440,8 +1440,9 @@ var _ = ginkgo.Describe("MultiKueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
 				acs := admissioncheck.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(multiKueueAC.Name))
 				g.Expect(acs).To(gomega.BeComparableTo(&kueue.AdmissionCheckState{
-					Name:  kueue.AdmissionCheckReference(multiKueueAC.Name),
-					State: kueue.CheckStatePending,
+					Name:       kueue.AdmissionCheckReference(multiKueueAC.Name),
+					State:      kueue.CheckStatePending,
+					RetryCount: ptr.To(int32(1)),
 				}, cmpopts.IgnoreFields(kueue.AdmissionCheckState{}, "LastTransitionTime", "Message")))
 
 				// The transition interval should be close to testingKeepReadyTimeout (taking into account the resolution of the LastTransitionTime field)
diff --git a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
index cc2af7ede..f9008fce0 100644
--- a/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
+++ b/test/integration/singlecluster/controller/admissionchecks/provisioning/provisioning_test.go
@@ -28,6 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	autoscaling "k8s.io/autoscaler/cluster-autoscaler/apis/provisioningrequest/autoscaling.x-k8s.io/v1"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
@@ -920,8 +921,11 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 			ginkgo.By("Checking the AdmissionCheck is set to Retry, and the workload has requeueState set", func() {
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
+					check := admissioncheck.FindAdmissionCheck(updatedWl.Status.AdmissionChecks, kueue.AdmissionCheckReference(ac.Name))
+					g.Expect(check.RequeueAfterSeconds).ToNot(gomega.BeNil())
+					g.Expect(check.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*check.RetryCount).To(gomega.Equal(int32(1)))
 					g.Expect(updatedWl.Status.RequeueState).NotTo(gomega.BeNil())
-					g.Expect(*updatedWl.Status.RequeueState.Count).To(gomega.Equal(int32(1)))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -1018,7 +1022,10 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
 					g.Expect(updatedWl.Status.RequeueState).NotTo(gomega.BeNil())
-					g.Expect(*updatedWl.Status.RequeueState.Count).To(gomega.Equal(int32(1)))
+					check := admissioncheck.FindAdmissionCheck(updatedWl.Status.AdmissionChecks, kueue.AdmissionCheckReference(ac.Name))
+					g.Expect(check.RequeueAfterSeconds).ToNot(gomega.BeNil())
+					g.Expect(check.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*check.RetryCount).To(gomega.Equal(int32(1)))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -1086,11 +1093,12 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 
 					g.Expect(updatedWl.Status.AdmissionChecks).To(gomega.ContainElement(gomega.BeComparableTo(
 						kueue.AdmissionCheckState{
-							Name:    kueue.AdmissionCheckReference(ac.Name),
-							State:   kueue.CheckStatePending,
-							Message: "Reset to Pending after eviction. Previously: Rejected",
+							Name:       kueue.AdmissionCheckReference(ac.Name),
+							State:      kueue.CheckStatePending,
+							Message:    "Reset to Pending after eviction. Previously: Rejected",
+							RetryCount: ptr.To(int32(1)),
 						},
-						cmpopts.IgnoreFields(kueue.AdmissionCheckState{}, "LastTransitionTime", "PodSetUpdates"))))
+						cmpopts.IgnoreFields(kueue.AdmissionCheckState{}, "LastTransitionTime", "PodSetUpdates", "RequeueAfterSeconds"))))
 					g.Expect(workload.IsActive(&updatedWl)).To(gomega.BeFalse())
 
 					ok, err := utiltesting.HasEventAppeared(ctx, k8sClient, corev1.Event{
@@ -1136,7 +1144,10 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
 					g.Expect(updatedWl.Status.RequeueState).NotTo(gomega.BeNil())
-					g.Expect(*updatedWl.Status.RequeueState.Count).To(gomega.Equal(int32(1)))
+					check := admissioncheck.FindAdmissionCheck(updatedWl.Status.AdmissionChecks, kueue.AdmissionCheckReference(ac.Name))
+					g.Expect(check.RequeueAfterSeconds).ToNot(gomega.BeNil())
+					g.Expect(check.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*check.RetryCount).To(gomega.Equal(int32(1)))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -1302,7 +1313,10 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
 					g.Expect(updatedWl.Status.RequeueState).NotTo(gomega.BeNil())
-					g.Expect(*updatedWl.Status.RequeueState.Count).To(gomega.Equal(int32(1)))
+					check := admissioncheck.FindAdmissionCheck(updatedWl.Status.AdmissionChecks, kueue.AdmissionCheckReference(ac.Name))
+					g.Expect(check.RequeueAfterSeconds).ToNot(gomega.BeNil())
+					g.Expect(check.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*check.RetryCount).To(gomega.Equal(int32(1)))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
@@ -1348,7 +1362,10 @@ var _ = ginkgo.Describe("Provisioning", ginkgo.Ordered, ginkgo.ContinueOnFailure
 				gomega.Eventually(func(g gomega.Gomega) {
 					g.Expect(k8sClient.Get(ctx, wlKey, &updatedWl)).To(gomega.Succeed())
 					g.Expect(updatedWl.Status.RequeueState).NotTo(gomega.BeNil())
-					g.Expect(*updatedWl.Status.RequeueState.Count).To(gomega.Equal(int32(2)))
+					check := admissioncheck.FindAdmissionCheck(updatedWl.Status.AdmissionChecks, kueue.AdmissionCheckReference(ac.Name))
+					g.Expect(check.RequeueAfterSeconds).ToNot(gomega.BeNil())
+					g.Expect(check.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*check.RetryCount).To(gomega.Equal(int32(2)))
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
diff --git a/test/integration/singlecluster/scheduler/delayedadmission/delayed_admission_test.go b/test/integration/singlecluster/scheduler/delayedadmission/delayed_admission_test.go
new file mode 100644
index 000000000..ad0b007fa
--- /dev/null
+++ b/test/integration/singlecluster/scheduler/delayedadmission/delayed_admission_test.go
@@ -0,0 +1,215 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package delayedadmission
+
+import (
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta2"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta2"
+	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
+	utiltestingapi "sigs.k8s.io/kueue/pkg/util/testing/v1beta2"
+	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/test/integration/framework"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var _ = ginkgo.Describe("SchedulerWithDelayedAdmissionChecks", func() {
+	var (
+		// Values referenced by tests:
+		defaultFlavor *kueue.ResourceFlavor
+		ns            *corev1.Namespace
+		clusterQ      *kueue.ClusterQueue
+		localQ        *kueue.LocalQueue
+		delayedCheck  *kueue.AdmissionCheck
+		realClock     = util.RealClock
+	)
+
+	ginkgo.JustBeforeEach(func() {
+		fwk.StartManager(ctx, cfg, managerAndSchedulerSetup(&configapi.Configuration{}))
+
+		defaultFlavor = utiltestingapi.MakeResourceFlavor("default").Obj()
+		util.MustCreate(ctx, k8sClient, defaultFlavor)
+
+		ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "delayed-retry-")
+
+		delayedCheck = utiltestingapi.MakeAdmissionCheck("delayed-check").ControllerName("ctrl").Obj()
+		util.MustCreate(ctx, k8sClient, delayedCheck)
+		util.SetAdmissionCheckActive(ctx, k8sClient, delayedCheck, metav1.ConditionTrue)
+
+		clusterQ = utiltestingapi.MakeClusterQueue("dev-cq").
+			ResourceGroup(*utiltestingapi.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "5").Obj()).
+			AdmissionChecks(kueue.AdmissionCheckReference(delayedCheck.Name)).
+			Obj()
+		util.MustCreate(ctx, k8sClient, clusterQ)
+
+		localQ = utiltestingapi.MakeLocalQueue("dev-queue", ns.Name).ClusterQueue(clusterQ.Name).Obj()
+		util.MustCreate(ctx, k8sClient, localQ)
+
+		util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQ)
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQ, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, defaultFlavor, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, delayedCheck, true)
+		fwk.StopManager(ctx)
+	})
+
+	ginkgo.Context("Delayed admission check", func() {
+		ginkgo.It("should requeue workloads after a delayed retry", framework.SlowSpec, func() {
+			wl := utiltestingapi.MakeWorkload("delayed-ac-retry", ns.Name).
+				RequestAndLimit(corev1.ResourceCPU, "1").
+				Queue(kueue.LocalQueueName(localQ.Name)).
+				Obj()
+
+			ginkgo.By("Creating the job", func() {
+				util.MustCreate(ctx, k8sClient, wl)
+			})
+			wlLookupKey := client.ObjectKeyFromObject(wl)
+			createdWorkload := &kueue.Workload{}
+
+			ginkgo.By("Verifying workload is created but not admitted yet", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					g.Expect(workload.HasQuotaReservation(createdWorkload)).To(gomega.BeTrue())
+					g.Expect(workload.IsAdmitted(createdWorkload)).To(gomega.BeFalse())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Marking AC as Retry to trigger an immediate retry", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					workload.SetAdmissionCheckState(&createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckState{
+						Name:    kueue.AdmissionCheckReference(delayedCheck.Name),
+						State:   kueue.CheckStateRetry,
+						Message: "Retrying admission check",
+					}, realClock)
+					g.Expect(k8sClient.Status().Update(ctx, createdWorkload)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Finish eviction", func() {
+				util.FinishEvictionForWorkloads(ctx, k8sClient, createdWorkload)
+			})
+
+			ginkgo.By("Verifying retry counter is incremented after transition to Pending", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+
+					ac := admissioncheck.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(delayedCheck.Name))
+					g.Expect(ac).ToNot(gomega.BeNil())
+					g.Expect(ac.State).To(gomega.Equal(kueue.CheckStatePending))
+					g.Expect(ac.LastTransitionTime).ToNot(gomega.BeNil())
+					g.Expect(ac.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*ac.RetryCount).To(gomega.Equal(int32(1)))
+
+					g.Expect(createdWorkload.Status.RequeueState).To(gomega.BeNil())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Wait for job to have quota again", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					g.Expect(workload.HasQuotaReservation(createdWorkload)).To(gomega.BeTrue())
+					g.Expect(workload.IsAdmitted(createdWorkload)).To(gomega.BeFalse())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Marking AC as Retry to trigger a delayed retry", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					workload.SetAdmissionCheckState(&createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckState{
+						Name:    kueue.AdmissionCheckReference(delayedCheck.Name),
+						State:   kueue.CheckStateRetry,
+						Message: "Retrying admission check",
+						// The RequeueState.RequeueAt timestamp has second-level precision, so to ensure the test passes,
+						// we set the timeout to five seconds.
+						// This is necessary because there's a race condition between the actual timeout and the test
+						// proceeding with its assertions. In the worst case, Kueue may reset the RequeueAt field just before
+						// we check if it’s set, causing the test to fail.
+						// To avoid flaky tests—especially in our CI environment—we've chosen to use a slightly longer delay.
+						RequeueAfterSeconds: ptr.To(int32(5)),
+					}, realClock)
+					g.Expect(k8sClient.Status().Update(ctx, createdWorkload)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Finish eviction", func() {
+				util.FinishEvictionForWorkloads(ctx, k8sClient, createdWorkload)
+			})
+
+			ginkgo.By("Verifying retry counter is incremented after transition to Pending", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+
+					ac := admissioncheck.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(delayedCheck.Name))
+					g.Expect(ac).ToNot(gomega.BeNil())
+					g.Expect(ac.State).To(gomega.Equal(kueue.CheckStatePending))
+					g.Expect(ac.LastTransitionTime).To(gomega.Not(gomega.BeNil()))
+					g.Expect(ac.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*ac.RetryCount).To(gomega.Equal(int32(2)))
+					g.Expect(ac.RequeueAfterSeconds).ToNot(gomega.BeNil())
+
+					g.Expect(createdWorkload.Status.RequeueState).ToNot(gomega.BeNil())
+					g.Expect(createdWorkload.Status.RequeueState.RequeueAt).ToNot(gomega.BeNil())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Wait for job to have quota again", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					g.Expect(workload.HasQuotaReservation(createdWorkload)).To(gomega.BeTrue())
+					g.Expect(workload.IsAdmitted(createdWorkload)).To(gomega.BeFalse())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Marking AC as Ready", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+					wlPatch := workload.BaseSSAWorkload(createdWorkload, true)
+					workload.SetAdmissionCheckState(&wlPatch.Status.AdmissionChecks, kueue.AdmissionCheckState{
+						Name:  kueue.AdmissionCheckReference(delayedCheck.Name),
+						State: kueue.CheckStateReady,
+					}, realClock)
+					g.Expect(k8sClient.Status().Patch(ctx, wlPatch, client.Apply, client.FieldOwner(kueue.MultiKueueControllerName), client.ForceOwnership)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Verifying workload is admitted with retry counter", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+
+					ac := admissioncheck.FindAdmissionCheck(createdWorkload.Status.AdmissionChecks, kueue.AdmissionCheckReference(delayedCheck.Name))
+					g.Expect(ac.State).To(gomega.Equal(kueue.CheckStateReady))
+					g.Expect(ac.LastTransitionTime).To(gomega.Not(gomega.BeNil()))
+					g.Expect(ac.RetryCount).ToNot(gomega.BeNil())
+					g.Expect(*ac.RetryCount).To(gomega.Equal(int32(2)))
+
+					g.Expect(workload.HasQuotaReservation(createdWorkload)).To(gomega.BeTrue())
+					g.Expect(workload.IsAdmitted(createdWorkload)).To(gomega.BeTrue())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+})
diff --git a/test/integration/singlecluster/scheduler/delayedadmission/suite_test.go b/test/integration/singlecluster/scheduler/delayedadmission/suite_test.go
new file mode 100644
index 000000000..ee6417c0d
--- /dev/null
+++ b/test/integration/singlecluster/scheduler/delayedadmission/suite_test.go
@@ -0,0 +1,102 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package delayedadmission
+
+import (
+	"context"
+	"testing"
+
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta2"
+	qcache "sigs.k8s.io/kueue/pkg/cache/queue"
+	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
+	"sigs.k8s.io/kueue/pkg/constants"
+	"sigs.k8s.io/kueue/pkg/controller/core"
+	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
+	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
+	"sigs.k8s.io/kueue/pkg/scheduler"
+	"sigs.k8s.io/kueue/pkg/webhooks"
+	"sigs.k8s.io/kueue/test/integration/framework"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var (
+	cfg       *rest.Config
+	k8sClient client.Client
+	ctx       context.Context
+	fwk       *framework.Framework
+)
+
+func TestSchedulerWithWaitForPodsReady(t *testing.T) {
+	gomega.RegisterFailHandler(ginkgo.Fail)
+
+	ginkgo.RunSpecs(t,
+		"Scheduler with delayed admission checks Suite",
+	)
+}
+
+var _ = ginkgo.BeforeSuite(func() {
+	fwk = &framework.Framework{
+		WebhookPath: util.WebhookPath,
+	}
+	cfg = fwk.Init()
+	ctx, k8sClient = fwk.SetupClient(cfg)
+})
+
+var _ = ginkgo.AfterSuite(func() {
+	fwk.Teardown()
+})
+
+func managerAndSchedulerSetup(configuration *configapi.Configuration) framework.ManagerSetup {
+	if configuration == nil {
+		configuration = &configapi.Configuration{}
+	}
+	return func(ctx context.Context, mgr manager.Manager) {
+		var (
+			cacheOpts  []schdcache.Option
+			queuesOpts []qcache.Option
+			schedOpts  []scheduler.Option
+		)
+
+		mgr.GetScheme().Default(configuration)
+
+		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		cCache := schdcache.New(mgr.GetClient(), cacheOpts...)
+		queues := qcache.NewManager(mgr.GetClient(), cCache, queuesOpts...)
+
+		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, configuration)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
+
+		failedWebhook, err := webhooks.Setup(mgr)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
+
+		err = workloadjob.SetupIndexes(ctx, mgr.GetFieldIndexer())
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+		sched := scheduler.New(queues, cCache, mgr.GetClient(), mgr.GetEventRecorderFor(constants.AdmissionName), schedOpts...)
+
+		err = sched.Start(ctx)
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+	}
+}
diff --git a/test/integration/singlecluster/scheduler/podsready/scheduler_test.go b/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
index 9f6078879..266ac0671 100644
--- a/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
+++ b/test/integration/singlecluster/scheduler/podsready/scheduler_test.go
@@ -778,15 +778,29 @@ var _ = ginkgo.Describe("SchedulerWithWaitForPodsReadyNonblockingMode", func() {
 				util.ExpectWorkloadsToHaveQuotaReservation(ctx, k8sClient, standaloneClusterQ.Name, wl1)
 				// To avoid flakiness, we don't verify if the workload has a QuotaReserved=false with pending reason here.
 			})
-			ginkgo.By("verifying if all workloads have a proper re-queue count", func() {
+			ginkgo.By("verifying requeue state after first eviction and re-admission", func() {
+				// After the first eviction and re-admission (backoff completed), wl1 should have:
+				// - Count=1 (from the first eviction)
+				// - RequeueAt=nil (cleared after re-admission in nonblocking mode)
+				util.ExpectWorkloadToHaveRequeueState(ctx, k8sClient, client.ObjectKeyFromObject(wl1), &kueue.RequeueState{
+					Count: ptr.To[int32](1),
+				}, false)
+			})
+			ginkgo.By("waiting for the first workload to be evicted again (second eviction)", func() {
+				util.AwaitWorkloadEvictionByPodsReadyTimeout(ctx, k8sClient, client.ObjectKeyFromObject(wl1), podsReadyTimeout)
+				util.FinishEvictionForWorkloads(ctx, k8sClient, wl1)
+			})
+			ginkgo.By("verifying if all workloads have a proper re-queue count after second eviction", func() {
 				// Here, we focus on verifying if the requeuingTimestamp works well.
-				// So, we don't check if the .status.requeueState.requeueAt is reset.
+				// We verify that .status.requeueState.requeueAt is nil (hasRequeueAt=false)
+				// because in nonblocking mode, after each backoff completes and re-admission happens,
+				// the controller clears RequeueAt to nil while preserving Count.
 				util.ExpectWorkloadToHaveRequeueState(ctx, k8sClient, client.ObjectKeyFromObject(wl1), &kueue.RequeueState{
 					Count: ptr.To[int32](2),
-				}, true)
+				}, false)
 				util.ExpectWorkloadToHaveRequeueState(ctx, k8sClient, client.ObjectKeyFromObject(wl2), &kueue.RequeueState{
 					Count: ptr.To[int32](1),
-				}, true)
+				}, false)
 				ginkgo.By("wl3 had never been admitted", func() {
 					util.ExpectWorkloadToHaveRequeueState(ctx, k8sClient, client.ObjectKeyFromObject(wl3), nil, false)
 				})