kaxil opened a new pull request, #74719:
URL: https://github.com/apache/airflow/pull/74719
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`.
```python
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`.
## 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
