[kaxil]

Closes #74660

A retry policy can decide RETRY, FAIL or DEFAULT, but some failures mean "not yet" rather than "this went wrong". A KubernetesPodOperator task whose namespace ResourceQuota is full gets a 403 on pod creation; at the start of the day that can last hours. Today the choice is to subclass the operator and raise AirflowRescheduleException, or to set dozens of retries, which then fire on_retry_callback every few minutes and leave no retries for a real failure once capacity frees up.

This adds RetryAction.RESCHEDULE. The task goes to up_for_reschedule, the state a sensor in reschedule mode already uses, so the try number and retries are untouched and on_retry_callback does not fire. It works with retries=0.

RetryRule(
    exception=WarehouseBusyError,
    action=RetryAction.RESCHEDULE,
    retry_delay=timedelta(minutes=10),
    reschedule_timeout=timedelta(hours=4),  # and/or max_reschedules=24
    reason="Warehouse busy",
)

# or from a custom policy
RetryDecision.reschedule(timedelta(minutes=10), reschedule_timeout=timedelta(hours=4), reason="Namespace quota full")

Design rationale

No Execution API or database change. Any task can already move to up_for_reschedule: the worker sends the same RescheduleTask message a sensor does, and the server and scheduler paths are unchanged. The policy's reason is logged rather than stored; storing it would need a new field on the reschedule payload.

Every RESCHEDULE is bounded. A rule matching a broad exception would otherwise turn every real failure into an endless loop that never touches retries. max_reschedules and reschedule_timeout live on RetryDecision and RetryRule, a RESCHEDULE without either raises ValueError when it is created (so a bad rule fails at parse time), and the task runner enforces them rather than trusting each policy to. Past a bound, the failure goes through the standard retry logic with a retry_reason such as Reschedule limit reached (reschedule_timeout=4:00:00): Warehouse busy. Setting either bound on another action also raises.

Bounds count the try's reschedules, the records sensors already use. task_reschedule_count and the first reschedule date reset when a new try starts, so a retry gets a fresh allowance and retries still caps the total.

This is for an external system the task drives, not for disruption of the Airflow worker. The policy runs in the worker and only sees the exception the task raised. A preempted worker pod is the executor's business (the subject of the AIP-97 classification and AIP-122 infrastructure retry proposals), and a policy never sees it.

Composing policies keeps the bounds. ChainRetryPolicy, its common.compat backport and ClassifierRetryPolicy's fallback paths rebuilt the winning decision field by field, which would drop the new fields and turn a valid RESCHEDULE into a ValueError. They now use dataclasses.replace.

A rescheduled failure still fails closed. If the terminal-state message cannot reach the supervisor, a task whose exception was turned into a reschedule now exits non-zero like FAILED and UP_FOR_RETRY do, so it is not reported as SUCCESS. A sensor's own reschedule is unchanged.

Run end to end with LocalExecutor and a task that always raises (max_reschedules=2, retries=1, retry_delay=5s, 10s reschedule delay): try 1 ran three times (two reschedules, then the bound), retried after the task's own 5s delay, try 2 ran three times the same way and then failed. task_reschedule held two rows per try, and both tries stored retry_reason = Reschedule limit reached (max_reschedules=2): cluster full.

The same Dag in the UI, raising a quota error with a 20s reschedule delay (the needs_capacity row is from an earlier version of the Dag). During try 1, after its second reschedule; both reschedules are in the same try's log:

After the bound on try 2, with the stored reason in the header:

Known issues

• Reschedules from any cause share the try's counter: a sensor's pokes and the worker's own reschedule of a task whose Dag is not yet visible both count toward the bounds, and policy reschedules use up [workers] missing_dag_retries. Sensors in reschedule mode already have this interaction. Telling them apart needs a source column on task_reschedule.
• The OpenLineage listener only skips a repeated START for operators with reschedule=True, so a policy-driven reschedule emits another START for the same run.
• Released apache-airflow-providers-common-ai versions rebuild decisions field by field, so a RESCHEDULE from ClassifierRetryPolicy's fallback_policy or fallback_rules falls back to DEFAULT until the provider ships this change.

[pankajastro]

A few questions inline.

---

Drafted-by: Claude Code (Sonnet 5.5); reviewed by @pankajastro before posting

[github-actions]

Backport failed to create: v3-4-test. View the failure log Run details

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
❌	v3-4-test

You can attempt to backport this manually by running:

cherry_picker 7c2c922 v3-4-test

This should apply the commit to the v3-4-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.