From 6778b1bb5856aa74cd2add21c96f1661f9e63bcd Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Mon, 15 Jun 2026 10:38:43 +0800 Subject: [PATCH 1/2] Block retry_policy argument for StubOperator --- .../src/airflow/providers/standard/decorators/stub.py | 10 ++++++++++ .../tests/unit/standard/decorators/test_stub.py | 11 +++++++++++ 2 files changed, 21 insertions(+) diff --git a/providers/standard/src/airflow/providers/standard/decorators/stub.py b/providers/standard/src/airflow/providers/standard/decorators/stub.py index f29d123c740c1..d5d9cd090d683 100644 --- a/providers/standard/src/airflow/providers/standard/decorators/stub.py +++ b/providers/standard/src/airflow/providers/standard/decorators/stub.py @@ -46,6 +46,16 @@ def __init__( task_id=task_id, **kwargs, ) + # A retry_policy is user Python evaluated in-process by the task runner. Stub tasks + # execute on a remote/native worker via the Task Execution Interface and never run the + # Python task runner, so the policy would silently never fire. Reject it up front. + # (retries is fine -- the server computes retry eligibility regardless of runtime.) + if self.retry_policy is not None: + raise ValueError( + "@task.stub does not support `retry_policy`: it runs Python in-process, but stub " + "tasks execute on a lang-sdk runtime and never evaluate the policy. Use `retries` " + "instead." + ) # Validate python callable module = ast.parse(self.get_python_source()) diff --git a/providers/standard/tests/unit/standard/decorators/test_stub.py b/providers/standard/tests/unit/standard/decorators/test_stub.py index 406318ca2b913..22c661e697404 100644 --- a/providers/standard/tests/unit/standard/decorators/test_stub.py +++ b/providers/standard/tests/unit/standard/decorators/test_stub.py @@ -55,3 +55,14 @@ def fn_code(): def test_stub_signature(fn, error): with error: stub(fn)() + + +def test_stub_rejects_retry_policy(): + from airflow.sdk.definitions.retry_policy import ExceptionRetryPolicy + + with pytest.raises(ValueError, match="does not support retry_policy"): + stub(fn_pass, retry_policy=ExceptionRetryPolicy(rules=[]))() + + +def test_stub_allows_retries(): + stub(fn_pass, retries=5)() From 999e9c951657f8197f96d401e733ba89c924779f Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Mon, 15 Jun 2026 13:24:33 +0800 Subject: [PATCH 2/2] Guard retry_policy check for Airflow < 3.3 compatibility --- .../src/airflow/providers/standard/decorators/stub.py | 2 +- .../standard/tests/unit/standard/decorators/test_stub.py | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/providers/standard/src/airflow/providers/standard/decorators/stub.py b/providers/standard/src/airflow/providers/standard/decorators/stub.py index d5d9cd090d683..08bcf163a56ad 100644 --- a/providers/standard/src/airflow/providers/standard/decorators/stub.py +++ b/providers/standard/src/airflow/providers/standard/decorators/stub.py @@ -50,7 +50,7 @@ def __init__( # execute on a remote/native worker via the Task Execution Interface and never run the # Python task runner, so the policy would silently never fire. Reject it up front. # (retries is fine -- the server computes retry eligibility regardless of runtime.) - if self.retry_policy is not None: + if getattr(self, "retry_policy", None) is not None: raise ValueError( "@task.stub does not support `retry_policy`: it runs Python in-process, but stub " "tasks execute on a lang-sdk runtime and never evaluate the policy. Use `retries` " diff --git a/providers/standard/tests/unit/standard/decorators/test_stub.py b/providers/standard/tests/unit/standard/decorators/test_stub.py index 22c661e697404..2a17c3fdd82c1 100644 --- a/providers/standard/tests/unit/standard/decorators/test_stub.py +++ b/providers/standard/tests/unit/standard/decorators/test_stub.py @@ -22,6 +22,8 @@ from airflow.providers.standard.decorators.stub import stub +from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS + def fn_ellipsis(): ... @@ -57,10 +59,11 @@ def test_stub_signature(fn, error): stub(fn)() +@pytest.mark.skipif(not AIRFLOW_V_3_3_PLUS, reason="retry_policy added in Airflow 3.3") def test_stub_rejects_retry_policy(): from airflow.sdk.definitions.retry_policy import ExceptionRetryPolicy - with pytest.raises(ValueError, match="does not support retry_policy"): + with pytest.raises(ValueError, match="does not support `retry_policy`"): stub(fn_pass, retry_policy=ExceptionRetryPolicy(rules=[]))()