(airflow) branch main updated: Catch BrokenPipeError gracefully while servicing subprocess (#49128)

[as...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ash pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 22d4476ae97 Catch BrokenPipeError gracefully while servicing subprocess (#49128)
22d4476ae97 is described below

commit 22d4476ae97da00b0bec50cd7e2a637374a33bbb
Author: Amogh Desai <am...@gmail.com>
AuthorDate: Fri Apr 11 22:18:54 2025 +0530

    Catch BrokenPipeError gracefully while servicing subprocess (#49128)
---
 airflow-core/src/airflow/dag_processing/manager.py    | 7 ++++++-
 task-sdk/src/airflow/sdk/execution_time/supervisor.py | 7 ++++++-
 2 files changed, 12 insertions(+), 2 deletions(-)

diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py
index 7bc046cd663..c15cabf68b1 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -381,8 +381,13 @@ class DagFileProcessorManager(LoggingMixin):
         events = self.selector.select(timeout=timeout)
         for key, _ in events:
             socket_handler = key.data
-            need_more = socket_handler(key.fileobj)
 
+            # BrokenPipeError should be caught and treated as if the handler returned false, similar
+            # to EOF case
+            try:
+                need_more = socket_handler(key.fileobj)
+            except BrokenPipeError:
+                need_more = False
             if not need_more:
                 self.selector.unregister(key.fileobj)
                 key.fileobj.close()  # type: ignore[union-attr]
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index a48b49574e5..931d49e5592 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -653,7 +653,12 @@ class WatchedSubprocess:
             # If the subprocess writes "Hello, World!" to stdout:
             # - `socket_handler` reads and processes the message.
             # - If EOF is reached, the handler returns False to signal no more reads are expected.
-            need_more = socket_handler(key.fileobj)
+            # - BrokenPipeError should be caught and treated as if the handler returned false, similar
+            # to EOF case
+            try:
+                need_more = socket_handler(key.fileobj)
+            except BrokenPipeError:
+                need_more = False
 
             # If the handler signals that the file object is no longer needed (EOF, closed, etc.)
             # unregister it from the selector to stop monitoring; `wait()` blocks until all selectors