diff --git a/task-sdk/src/airflow/sdk/execution_time/timeout.py b/task-sdk/src/airflow/sdk/execution_time/timeout.py index 925916ed4f947..46032a27d0a79 100644 --- a/task-sdk/src/airflow/sdk/execution_time/timeout.py +++ b/task-sdk/src/airflow/sdk/execution_time/timeout.py @@ -16,6 +16,7 @@ # under the License. from __future__ import annotations +import contextlib import os import structlog @@ -35,7 +36,9 @@ def __init__(self, seconds=1, error_message="Timeout"): def handle_timeout(self, signum, frame): """Log information and raises AirflowTaskTimeout.""" - self.log.error("Process timed out", pid=os.getpid()) + # Ensure the timeout isn't masked by a logging error + with contextlib.suppress(OSError): + self.log.error("Process timed out", pid=os.getpid()) raise AirflowTaskTimeout(self.error_message) def __enter__(self): diff --git a/task-sdk/tests/task_sdk/execution_time/test_timeout.py b/task-sdk/tests/task_sdk/execution_time/test_timeout.py index 94f7babd77d07..99980b92fb3b6 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_timeout.py +++ b/task-sdk/tests/task_sdk/execution_time/test_timeout.py @@ -75,3 +75,29 @@ def test_timeout_happens(self): tp = TimeoutPosix(seconds=1, error_message="boom") with pytest.raises(AirflowTaskTimeout, match="boom"): tp.handle_timeout(signal.SIGALRM, None) + + def test_handle_timeout_raises_even_when_log_write_fails(self): + if not hasattr(signal, "SIGALRM"): + pytest.skip("SIGALRM not supported on this platform") + + tp = TimeoutPosix(seconds=1, error_message="boom") + tp.log = mock.MagicMock() + tp.log.error.side_effect = BrokenPipeError("supervisor pipe closed") + + with pytest.raises(AirflowTaskTimeout, match="boom"): + tp.handle_timeout(signal.SIGALRM, None) + + tp.log.error.assert_called_once_with("Process timed out", pid=mock.ANY) + + def test_handle_timeout_normal_path_logs_and_raises(self): + """Regression guard: on the happy path the log is emitted AND AirflowTaskTimeout is raised.""" + if not hasattr(signal, "SIGALRM"): + pytest.skip("SIGALRM not supported on this platform") + + tp = TimeoutPosix(seconds=1, error_message="boom") + tp.log = mock.MagicMock() + + with pytest.raises(AirflowTaskTimeout, match="boom"): + tp.handle_timeout(signal.SIGALRM, None) + + tp.log.error.assert_called_once_with("Process timed out", pid=mock.ANY)