diff --git a/providers/standard/src/airflow/providers/standard/decorators/stub.py b/providers/standard/src/airflow/providers/standard/decorators/stub.py index f29d123c740c1..08bcf163a56ad 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 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` " + "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..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(): ... @@ -55,3 +57,15 @@ def fn_code(): def test_stub_signature(fn, error): with 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`"): + stub(fn_pass, retry_policy=ExceptionRetryPolicy(rules=[]))() + + +def test_stub_allows_retries(): + stub(fn_pass, retries=5)()