Preserve AirflowTaskTimeout when timeout logging fails

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,11 @@ 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())
+        # Logging must never mask the timeout: task-runner logs go through
+        # a pipe owned by the supervisor, and a broken pipe here would
+        # replace AirflowTaskTimeout with BrokenPipeError. See #64212.
+        with contextlib.suppress(OSError):
+            self.log.error("Process timed out", pid=os.getpid())
         raise AirflowTaskTimeout(self.error_message)
 
     def __enter__(self):

task-sdk/tests/task_sdk/execution_time/test_timeout.py

@@ -75,3 +75,35 @@ 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):
+        # Task-runner logs go through a pipe owned by the supervisor. If that
+        # pipe is broken when SIGALRM fires (e.g. supervisor already died),
+        # self.log.error raises BrokenPipeError inside the signal handler.
+        # Without a guard, that OSError replaces the AirflowTaskTimeout, so
+        # the task exits with the wrong exception class and skips the
+        # timeout-handling path in task_runner. See #64212.
+        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)