This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e63e6bb5244b4c62b8384ff9990f944ae5a92840
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Nov 4 02:17:17 2026 +0000
[v3-4-test] Fix retry policy docs and the example Dag's `ConnectionError`
rules (#76313) (#76406)
The example Dag's ConnectionError rules used the builtin ConnectionError,
which requests.exceptions.ConnectionError does not subclass, so they never
matched what the example task raises. The docs marked RESCHEDULE as added in
3.5.0 (it shipped in 3.4.0), quoted a chain reason the page's own policy cannot
produce, said every policy decision is logged (only decisions with a reason
are), and listed only AirflowFailException as skipping the policy.
(cherry picked from commit 1631eccfc2d72764bd0d3f15b4831ad25e031b51)
Co-authored-by: Kaxil Naik <[email protected]>
---
airflow-core/docs/core-concepts/tasks.rst | 33 ++++++++++++++--------
.../airflow/example_dags/example_retry_policy.py | 13 +++------
2 files changed, 26 insertions(+), 20 deletions(-)
diff --git a/airflow-core/docs/core-concepts/tasks.rst
b/airflow-core/docs/core-concepts/tasks.rst
index f57de76baa8..128ad8c2ba7 100644
--- a/airflow-core/docs/core-concepts/tasks.rst
+++ b/airflow-core/docs/core-concepts/tasks.rst
@@ -215,8 +215,11 @@ How it works
~~~~~~~~~~~~
The policy runs in the **task worker process** (never the scheduler) between
catching the
-exception and deciding the task's next state. Each policy decision is logged
in the task
-logs as ``Retry policy decision action=<action> reason=<reason>``.
+exception and deciding the task's next state. A decision that has a reason is
logged in the
+task logs as ``Retry policy decision action=<action> reason=<reason>``. A
matching
+``RetryRule`` always gives one (its ``reason``, or ``Matched rule for
<exception type>``), and
+so does ``ChainRetryPolicy``. When no rule matches, or a custom policy returns
a decision
+without a reason, no line is logged.
When a task fails, the policy evaluates the exception and returns one of four
actions:
@@ -263,15 +266,15 @@ Composition with existing parameters
A reschedule without its own delay waits ``retry_delay``, without
backoff.
* - ``on_retry_callback``
- Fires on retries only, including policy-driven retries.
- * - ``AirflowFailException``
- - Always takes precedence. The policy is never consulted for this
exception.
+ * - ``AirflowFailException``, ``AirflowSensorTimeout``,
``AirflowTaskTerminated``
+ - Always take precedence. The task fails without consulting the policy.
.. _concepts:retry-policies-reschedule:
Rescheduling without using a retry
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-.. versionadded:: 3.5.0
+.. versionadded:: 3.4.0
Some failures mean "not yet" rather than "this went wrong": the system the
task drives is
out of capacity, such as a Kubernetes namespace whose resource quota is full
or a warehouse
@@ -355,11 +358,15 @@ Mapped tasks
~~~~~~~~~~~~
Policies work with task mapping. The policy applies per mapped task
-instance -- if instance 2 of 10 hits FAIL, the other 9 continue independently:
+instance -- if instance 2 of 10 hits FAIL, the other 9 continue independently.
Using
+``STANDARD_RETRY_POLICY`` from the previous example:
.. code-block:: python
- @task(retry_policy=my_policy)
+ from airflow.sdk import task
+
+
+ @task(retry_policy=STANDARD_RETRY_POLICY)
def my_mapped_task(input): ...
@@ -369,7 +376,9 @@ For a classic operator, pass the policy to ``.partial()``:
.. code-block:: python
- BashOperator.partial(task_id="run", retry_policy=my_policy).expand(
+ from airflow.providers.standard.operators.bash import BashOperator
+
+ BashOperator.partial(task_id="run",
retry_policy=STANDARD_RETRY_POLICY).expand(
bash_command=["echo 1", "echo 2", "echo 3"],
)
@@ -409,9 +418,11 @@ Two consequences of "DEFAULT means next" to keep in mind:
A policy that raises an ordinary exception, or returns something other than a
``RetryDecision``, is logged and treated as DEFAULT, so one broken policy does
not take the
rules after it down with it. The winning decision's reason names the policy
that decided and
-then what the earlier ones said (``HTTPStatusRetryPolicy: HTTP 503 (after
ExceptionRetryPolicy:
-no decision)``). That string is the task's ``retry_reason`` on a FAIL as well
as a RETRY; when
-no policy decided, it appears only in the task log as the ``Retry policy
decision`` line.
+then what the earlier ones said, for example ``HTTPStatusRetryPolicy: HTTP 404
(after
+ExceptionRetryPolicy: no decision)``. A decision with no reason of its own
shows its action
+instead, such as ``HTTPStatusRetryPolicy: retry``. That string is the task's
``retry_reason``
+on a FAIL as well as a RETRY; when no policy decided, it appears only in the
task log as the
+``Retry policy decision`` line.
Custom retry policies
~~~~~~~~~~~~~~~~~~~~~
diff --git a/airflow-core/src/airflow/example_dags/example_retry_policy.py
b/airflow-core/src/airflow/example_dags/example_retry_policy.py
index 383c5082a68..546c2ca1759 100644
--- a/airflow-core/src/airflow/example_dags/example_retry_policy.py
+++ b/airflow-core/src/airflow/example_dags/example_retry_policy.py
@@ -24,9 +24,9 @@ using a retry, or fall through to the standard retry logic.
from __future__ import annotations
+# [START retry_policy_definition]
from datetime import timedelta
-# [START retry_policy_definition]
from airflow.sdk import DAG, ChainRetryPolicy, ExceptionRetryPolicy,
RetryAction, RetryRule, task
API_RETRY_POLICY = ExceptionRetryPolicy(
@@ -43,7 +43,7 @@ API_RETRY_POLICY = ExceptionRetryPolicy(
reason="Auth failure, not retryable",
),
RetryRule(
- exception=ConnectionError,
+ exception="requests.exceptions.ConnectionError",
action=RetryAction.RETRY,
retry_delay=timedelta(seconds=30),
),
@@ -52,12 +52,7 @@ API_RETRY_POLICY = ExceptionRetryPolicy(
# [END retry_policy_definition]
# [START retry_policy_on_task]
-with DAG(
- dag_id="example_retry_policy",
- schedule=None,
- catchup=False,
- tags=["example", "retry_policy"],
-):
+with DAG(dag_id="example_retry_policy", tags=["example", "retry_policy"]):
@task(retries=5, retry_delay=timedelta(minutes=1),
retry_policy=API_RETRY_POLICY)
def call_external_api():
@@ -75,7 +70,7 @@ with DAG(
STANDARD_RETRY_POLICY = ExceptionRetryPolicy(
rules=[
RetryRule(exception="requests.exceptions.HTTPError",
action=RetryAction.FAIL),
- RetryRule(exception=ConnectionError,
retry_delay=timedelta(seconds=10)),
+ RetryRule(exception="requests.exceptions.ConnectionError",
retry_delay=timedelta(seconds=10)),
],
)
# [END retry_policy_reusable]
