From 028148b065470e15a6135696c71c8902ffdec443 Mon Sep 17 00:00:00 2001 From: rjgoyln Date: Tue, 25 Aug 2026 18:35:53 +0800 Subject: [PATCH] Fail deferred sensors on non-timeout errors when soft_fail is set A sensor running poke() only skips on soft_fail when it times out or asks to fail outright; every other error fails the task, so a misconfigured sensor is visible. The same sensor in deferrable mode skipped on any error, including a crashed trigger, which left users unable to tell a sensor that legitimately gave up from one that never worked. closes: #71255 --- generated/known_airflow_exceptions.txt | 1 - task-sdk/src/airflow/sdk/bases/sensor.py | 30 +++++++----- task-sdk/tests/task_sdk/bases/test_sensor.py | 50 ++++++++++++-------- 3 files changed, 49 insertions(+), 32 deletions(-) diff --git a/generated/known_airflow_exceptions.txt b/generated/known_airflow_exceptions.txt index acfd9ae3eb825..34c75ddc69bfc 100644 --- a/generated/known_airflow_exceptions.txt +++ b/generated/known_airflow_exceptions.txt @@ -418,5 +418,4 @@ task-sdk/src/airflow/sdk/definitions/connection.py::4 task-sdk/src/airflow/sdk/definitions/decorators/setup_teardown.py::4 task-sdk/src/airflow/sdk/definitions/xcom_arg.py::1 task-sdk/src/airflow/sdk/execution_time/task_runner.py::1 -task-sdk/tests/task_sdk/bases/test_sensor.py::1 task-sdk/tests/task_sdk/execution_time/test_task_runner.py::2 diff --git a/task-sdk/src/airflow/sdk/bases/sensor.py b/task-sdk/src/airflow/sdk/bases/sensor.py index 3f1f0842e8982..bcbe66d9817f9 100644 --- a/task-sdk/src/airflow/sdk/bases/sensor.py +++ b/task-sdk/src/airflow/sdk/bases/sensor.py @@ -35,13 +35,16 @@ AirflowSensorTimeout, AirflowSkipException, AirflowTaskTimeout, - TaskDeferralError, TaskDeferralTimeout, ) if TYPE_CHECKING: from airflow.sdk.definitions.context import Context +# soft_fail turns these into a skip whether they surface from poke() or after resuming from a +# trigger; every other error fails the sensor. Both paths share the tuple so they cannot drift. +_SOFT_FAIL_EXCEPTIONS = (AirflowSensorTimeout, AirflowTaskTimeout, AirflowFailException) + class PokeReturnValue: """ @@ -69,8 +72,8 @@ class BaseSensorOperator(BaseOperator): Sensor operators keep executing at a time interval and succeed when a criteria is met and fail if and when they time out. - :param soft_fail: Set to true to mark the task as SKIPPED on failure. - Mutually exclusive with never_fail. + :param soft_fail: Set to true to mark the task as SKIPPED when it times out or raises + ``AirflowFailException``. Mutually exclusive with never_fail. :param poke_interval: Time that the job should wait in between each try. Can be ``timedelta`` or ``float`` seconds. :param timeout: Time elapsed before the task times out and fails. @@ -205,11 +208,7 @@ def run_duration() -> float: while True: try: poke_return = self.poke(context) - except ( - AirflowSensorTimeout, - AirflowTaskTimeout, - AirflowFailException, - ) as e: + except _SOFT_FAIL_EXCEPTIONS as e: if self.soft_fail: raise AirflowSkipException("Skipping due to soft_fail is set to True.") from e if self.never_fail: @@ -251,19 +250,28 @@ def run_duration() -> float: return xcom_value def resume_execution(self, next_method: str, next_kwargs: dict[str, Any] | None, context: Context): - # Use nested try/except to convert TaskDeferralTimeout to AirflowSensorTimeout - # while still allowing soft_fail/never_fail to handle both exception types. + # Nested try/except so a trigger timeout becomes AirflowSensorTimeout before soft_fail + # and never_fail are applied to it. try: try: return super().resume_execution(next_method, next_kwargs, context) except TaskDeferralTimeout as e: raise AirflowSensorTimeout(*e.args) from e - except (AirflowException, TaskDeferralError) as e: + except _SOFT_FAIL_EXCEPTIONS as e: if self.soft_fail: raise AirflowSkipException("Skipping due to soft_fail is set to True.") from e if self.never_fail: raise AirflowSkipException("Skipping due to never_fail is set to True.") from e raise + except AirflowSkipException: + raise + except AirflowException as e: + # execute() only skips the exceptions handled above; anything else here (a crashed + # trigger, an error event raised by execute_complete) is a real failure, and soft_fail + # must not hide it behind a skip. + if self.never_fail: + raise AirflowSkipException("Skipping due to never_fail is set to True.") from e + raise def _get_next_poke_interval( self, diff --git a/task-sdk/tests/task_sdk/bases/test_sensor.py b/task-sdk/tests/task_sdk/bases/test_sensor.py index 5e7f588a1551a..690197116e426 100644 --- a/task-sdk/tests/task_sdk/bases/test_sensor.py +++ b/task-sdk/tests/task_sdk/bases/test_sensor.py @@ -62,12 +62,13 @@ def poke(self, context: Context): class DummyAsyncSensor(BaseSensorOperator): - def __init__(self, return_value=False, **kwargs): + def __init__(self, return_value=False, failure_exception: BaseException | None = None, **kwargs): super().__init__(**kwargs) self.return_value = return_value + self.failure_exception = failure_exception or AirflowException("Sensor failed") def execute_complete(self, context, event=None): - raise AirflowException("Should be skipped") + raise self.failure_exception class DummySensorWithXcomValue(BaseSensorOperator): @@ -688,16 +689,25 @@ def test_poke_mode_only_bad_poke(self): class TestAsyncSensor: + @pytest.mark.parametrize("soft_fail", [True, False]) + def test_error_from_execute_complete_fails_regardless_of_soft_fail(self, soft_fail): + async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail) + with pytest.raises(AirflowException, match="Sensor failed"): + async_sensor.resume_execution("execute_complete", None, {}) + @pytest.mark.parametrize( - ("soft_fail", "expected_exception"), + "failure_exception", [ - (True, AirflowSkipException), - (False, AirflowException), + pytest.param(AirflowSensorTimeout("timed out"), id="sensor-timeout"), + pytest.param(AirflowTaskTimeout("timed out"), id="task-timeout"), + pytest.param(AirflowFailException("failed"), id="fail-exception"), ], ) - def test_fail_after_resuming_deferred_sensor(self, soft_fail, expected_exception): - async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail) - with pytest.raises(expected_exception): + def test_soft_fail_skips_timeout_and_fail_exceptions_after_resuming(self, failure_exception): + async_sensor = DummyAsyncSensor( + task_id="dummy_async_sensor", soft_fail=True, failure_exception=failure_exception + ) + with pytest.raises(AirflowSkipException): async_sensor.resume_execution("execute_complete", None, {}) @pytest.mark.parametrize( @@ -727,25 +737,25 @@ def test_timeout_after_resuming_deferred_sensor_with_never_fail(self): context={}, ) - @pytest.mark.parametrize( - ("soft_fail", "expected_exception"), - [ - (True, AirflowSkipException), - (False, TaskDeferralError), - ], - ) - def test_trigger_failure_after_resuming_deferred_sensor_with_soft_fail( - self, soft_fail, expected_exception - ): - """Test that deferrable sensors with soft_fail skip on trigger failure instead of failing.""" + @pytest.mark.parametrize("soft_fail", [True, False]) + def test_trigger_failure_after_resuming_deferred_sensor_fails_regardless_of_soft_fail(self, soft_fail): async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail) - with pytest.raises(expected_exception): + with pytest.raises(TaskDeferralError): async_sensor.resume_execution( next_method="__fail__", next_kwargs={"error": TriggerFailureReason.TRIGGER_FAILURE}, context={}, ) + def test_never_fail_keeps_the_skip_raised_by_execute_complete(self): + async_sensor = DummyAsyncSensor( + task_id="dummy_async_sensor", + never_fail=True, + failure_exception=AirflowSkipException("External job has skipped"), + ) + with pytest.raises(AirflowSkipException, match="External job has skipped"): + async_sensor.resume_execution("execute_complete", None, {}) + def test_trigger_failure_after_resuming_deferred_sensor_with_never_fail(self): """Test that deferrable sensors with never_fail skip on trigger failure.""" async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", never_fail=True)