sdaberdaku opened a new pull request, #74740:
URL: https://github.com/apache/airflow/pull/74740
Follow-up to AIP-105 (#65474) and `ChainRetryPolicy` (#73508). A retry
policy can now end a task as `skipped`.
## Summary
`RetryAction.SKIP` (and the `RetryDecision.skip(reason=...)` helper) marks
the task `SKIPPED` instead of failing or retrying it. No retry is consumed,
failure callbacks and failure metrics do not fire, `on_skipped_callback` and
the skipped listener hook run, and downstream tasks see a skipped upstream
exactly as for `AirflowSkipException`. The decision's `reason` is stored as
`retry_reason`, as it already is for `RETRY` and `FAIL`.
```python
from airflow.sdk import ExceptionRetryPolicy, RetryAction, RetryRule
retry_policy = ExceptionRetryPolicy(
rules=[
RetryRule(exception=NoNewDataError, action=RetryAction.SKIP,
reason="nothing to ingest"),
RetryRule(exception=ConnectionError, action=RetryAction.RETRY,
retry_delay=timedelta(seconds=30)),
RetryRule(exception=PermissionError, action=RetryAction.FAIL),
]
)
```
## Why
AIP-105 made the retry policy the one place that maps an exception to an
outcome, but that mapping stopped at "retry" and "fail". A domain exception
that means "nothing to do" (no input for this partition, upstream source
unchanged, feature flag off) still had to raise `AirflowSkipException` from
inside business code, coupling it to Airflow, or be translated by a
`try`/`except` in every task body. With `SKIP` the whole exception-to-outcome
mapping lives in one reusable policy object attached to the task.
`FAIL` already set the precedent that a policy decides the task's outcome
rather than only its retry count. `SKIP` is the same kind of decision.
## What changes
- `RetryAction.SKIP` and `RetryDecision.skip()` in
`airflow/sdk/definitions/retry_policy.py`.
- `_handle_current_task_failed` in `task_runner.py` returns the same
`TaskState(state=SKIPPED, ...)` the `AirflowSkipException` path returns,
carrying the policy reason. The existing `finalize` path handles callbacks and
listeners.
- `ChainRetryPolicy` already treats any non-`DEFAULT` decision as the
winner, so only its docstring changes.
- Docs: a `SKIP` entry in the retry-policy section of
`core-concepts/tasks.rst`, plus a rule in `example_retry_policy.py`.
No API server, serialization, migration or UI change.
`TITerminalStatePayload` already accepts `SKIPPED` and the route already
persists `retry_reason` for any terminal payload. `AirflowFailException` and
`AirflowSensorTimeout` keep bypassing the policy, as today.
## Tests
- `test_retry_policy.py`: `RetryDecision.skip()`, a `SKIP` rule surviving
`serialize()`/`deserialize()`, `ChainRetryPolicy` letting a `SKIP` win, and
`_handle_current_task_failed` returning `SKIPPED` with the reason even when
`should_retry` is true.
- `test_task_runner.py`: an end-to-end `run()` with a `SKIP` rule ends
`SKIPPED`, carries `retry_reason`, and emits no `ti_failures` metric.
Run locally: both task-sdk test files, core `always`, `serialization` and
Execution API tests touching retry policy and example Dags, `prek run
--from-ref main` (pre-commit and manual stages, including `mypy-task-sdk`).
`TestTriggerDagRunOperator` in `test_task_runner.py` fails on `main` too on a
non-UTC host and passes with `TZ=UTC`; unrelated to this change.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Fable 5.1)
Generated-by: Claude Code (Fable 5.1) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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]
