[sdaberdaku]

Follow-up to AIP-105 (#65474), ChainRetryPolicy (#73508) and RetryAction.RESCHEDULE (#74719). 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 as they do for AirflowSkipException. The decision's reason is stored as retry_reason, as it already is for RETRY and FAIL.

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

A task can raise one of three Airflow exceptions to end some way other than a plain failure: AirflowFailException, AirflowRescheduleException and AirflowSkipException. A retry policy can now choose the first two outcomes itself, with FAIL (AIP-105) and RESCHEDULE (#74719). SKIP adds the third, so every one of these outcomes can come from a domain exception without business code importing Airflow's.

#74719 made this case for reschedule: without it, the only way was to subclass the operator and raise AirflowRescheduleException. Skipping has the same problem today. A domain exception that means "nothing to do" (no input for this partition, upstream source unchanged, feature flag off) has to raise AirflowSkipException from inside business code, which ties that code to Airflow, or be caught 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.

How it differs from RESCHEDULE

• No bounds. RESCHEDULE needs max_reschedules or reschedule_timeout because a broad rule could otherwise loop forever. SKIP ends the task, so it cannot loop and adds no fields to RetryDecision or RetryRule. A broad SKIP rule can hide real bugs as skips, so the docs say to match narrow exception types.
• The reason is stored. RESCHEDULE only logs its reason because RescheduleTask has no field for it. SKIP sends a TaskState, which already carries retry_reason.
• A lost final state ends like a lost AirflowSkipException. If the final state never reaches the supervisor, the scheduler finds the task still running when the worker exits and fails it: retried if retries remain, otherwise failed. Downstream tasks do not run.

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 on the current main: both task-sdk test files (657 passed with TZ=UTC) and prek run --from-ref main --stage pre-commit, which includes mypy for task-sdk and airflow-core. Before rebasing onto #74719, the core always, serialization and Execution API tests that touch retry policies and example Dags, and the manual prek stage, also passed; those have not been re-run since. TestTriggerDagRunOperator in test_task_runner.py also fails on main when the host is not on UTC, and passes with TZ=UTC; it is unrelated to this change.

closes: #75796

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Fable 5.1, Opus 5.5)

Generated-by: Claude Code (Fable 5.1, Opus 5.5) following the guidelines

[sdaberdaku]

@pankajastro, since you reviewed the RESCHEDULE action in #74719, would you mind taking a look at this one when you have a moment? It adds RetryAction.SKIP in the same retry-policy code and is rebased on top of that change. Thanks!