Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions task-sdk/src/airflow/sdk/bases/sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -251,13 +251,18 @@ 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.
try:
return super().resume_execution(next_method, next_kwargs, context)
except TaskDeferralTimeout as e:
raise AirflowSensorTimeout(*e.args) from e
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:
if self.soft_fail:
raise AirflowSkipException(str(e)) from e
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

def _get_next_poke_interval(
Expand Down
57 changes: 57 additions & 0 deletions task-sdk/tests/task_sdk/bases/test_sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
AirflowSensorTimeout,
AirflowSkipException,
AirflowTaskTimeout,
TaskDeferralError,
)
from airflow.models.trigger import TriggerFailureReason
from airflow.providers.standard.operators.empty import EmptyOperator
Expand Down Expand Up @@ -677,3 +678,59 @@ 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):
async_sensor.resume_execution("execute_complete", None, {})

@pytest.mark.parametrize(
("soft_fail", "expected_exception"),
[
(True, AirflowSkipException),
(False, AirflowSensorTimeout),
],
)
def test_timeout_after_resuming_deferred_sensor_with_soft_fail(self, soft_fail, expected_exception):
"""Test that deferrable sensors with soft_fail skip on timeout instead of failing."""
async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail)
with pytest.raises(expected_exception):
async_sensor.resume_execution(
next_method="__fail__",
next_kwargs={"error": TriggerFailureReason.TRIGGER_TIMEOUT},
context={},
)

def test_timeout_after_resuming_deferred_sensor_with_never_fail(self):
"""Test that deferrable sensors with never_fail skip on timeout."""
async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", never_fail=True)
with pytest.raises(AirflowSkipException):
async_sensor.resume_execution(
next_method="__fail__",
next_kwargs={"error": TriggerFailureReason.TRIGGER_TIMEOUT},
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."""
async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail)
with pytest.raises(expected_exception):
async_sensor.resume_execution(
next_method="__fail__",
next_kwargs={"error": TriggerFailureReason.TRIGGER_FAILURE},
context={},
)

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)
with pytest.raises(AirflowSkipException):
async_sensor.resume_execution(
next_method="__fail__",
next_kwargs={"error": TriggerFailureReason.TRIGGER_FAILURE},
context={},
)