This is an automated email from the ASF dual-hosted git repository.
shahar1 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 9ad8d78c277 Close SubprocessHook stdout pipe after command completes
(#74780)
9ad8d78c277 is described below
commit 9ad8d78c27779b3cc6660d1a496676a4a195c203
Author: Dev-iL <[email protected]>
AuthorDate: Sun Oct 18 17:34:18 2026 +0300
Close SubprocessHook stdout pipe after command completes (#74780)
Release the stdout pipe after commands finish or output reading fails,
avoiding leaked descriptors and ResourceWarnings. Keep it open on timeout
so the child can write while handling SIGTERM.
---
.../airflow/providers/standard/hooks/subprocess.py | 28 +++++++---
.../tests/unit/standard/hooks/test_subprocess.py | 63 +++++++++++++++++++++-
2 files changed, 84 insertions(+), 7 deletions(-)
diff --git
a/providers/standard/src/airflow/providers/standard/hooks/subprocess.py
b/providers/standard/src/airflow/providers/standard/hooks/subprocess.py
index 11626fb451a..3251714f0f1 100644
--- a/providers/standard/src/airflow/providers/standard/hooks/subprocess.py
+++ b/providers/standard/src/airflow/providers/standard/hooks/subprocess.py
@@ -93,18 +93,34 @@ class SubprocessHook(BaseHook):
line = ""
if self.sub_process is None:
raise RuntimeError("The subprocess should be created here and
is None!")
- if self.sub_process.stdout is not None:
- for raw_line in iter(self.sub_process.stdout.readline, b""):
- line = raw_line.decode(output_encoding,
errors="backslashreplace").rstrip()
- self.log.info("%s", line)
-
- self.sub_process.wait()
+ # Neither `with Popen(...)` nor a `finally` close, because both
act on the
+ # execution_timeout path, where AirflowTaskTimeout (a
BaseException) unwinds through here
+ # and on_kill() sends SIGTERM to the child only afterwards:
__exit__ would block in
+ # wait() so that SIGTERM never arrives, and closing stdout would
SIGPIPE a child that
+ # writes during its own TERM cleanup, since restore_signals=True
leaves SIGPIPE at its
+ # default action. Closing on success and on Exception leaves that
path as it was.
+ try:
+ if self.sub_process.stdout is not None:
+ for raw_line in iter(self.sub_process.stdout.readline,
b""):
+ line = raw_line.decode(output_encoding,
errors="backslashreplace").rstrip()
+ self.log.info("%s", line)
+
+ self.sub_process.wait()
+ except Exception:
+ self._close_stdout()
+ raise
+ else:
+ self._close_stdout()
self.log.info("Command exited with return code %s",
self.sub_process.returncode)
return_code: int = self.sub_process.returncode
return SubprocessResult(exit_code=return_code, output=line)
+ def _close_stdout(self) -> None:
+ if self.sub_process is not None and self.sub_process.stdout is not
None:
+ self.sub_process.stdout.close()
+
def send_sigterm(self):
"""Send SIGTERM signal to ``self.sub_process`` if one exists."""
self.log.info("Sending SIGTERM signal to process group")
diff --git a/providers/standard/tests/unit/standard/hooks/test_subprocess.py
b/providers/standard/tests/unit/standard/hooks/test_subprocess.py
index 044af2c1419..7eaee37675c 100644
--- a/providers/standard/tests/unit/standard/hooks/test_subprocess.py
+++ b/providers/standard/tests/unit/standard/hooks/test_subprocess.py
@@ -18,8 +18,9 @@
from __future__ import annotations
import signal
+from io import BufferedReader
from pathlib import Path
-from subprocess import PIPE, STDOUT
+from subprocess import PIPE, STDOUT, Popen
from tempfile import TemporaryDirectory
from unittest import mock
from unittest.mock import MagicMock
@@ -32,6 +33,10 @@ OS_ENV_KEY = "SUBPROCESS_ENV_TEST"
OS_ENV_VAL = "this-is-from-os-environ"
+class _TaskTimeout(BaseException):
+ """Stands in for ``AirflowTaskTimeout``, which is a ``BaseException`` so
task code cannot catch it."""
+
+
class TestSubprocessHook:
@pytest.mark.parametrize(
("env", "expected"),
@@ -103,6 +108,62 @@ class TestSubprocessHook:
stdout=PIPE,
)
+ def test_run_command_closes_stdout_pipe(self):
+ hook = SubprocessHook()
+ hook.run_command(command=["bash", "-c", 'echo "stdout"'])
+ assert hook.sub_process.stdout.closed
+
+ @mock.patch(
+ "airflow.providers.standard.hooks.subprocess.Popen",
+ return_value=MagicMock(
+ spec=Popen,
+ stdout=MagicMock(spec=BufferedReader,
readline=MagicMock(side_effect=OSError("pipe broken"))),
+ ),
+ )
+ def test_run_command_closes_stdout_pipe_on_read_error(self, mock_popen):
+ hook = SubprocessHook()
+ with pytest.raises(OSError, match="pipe broken"):
+ hook.run_command(command=["bash", "-c", 'echo "stdout"'])
+ mock_popen.return_value.stdout.close.assert_called_once()
+ mock_popen.return_value.wait.assert_not_called()
+
+ def test_run_command_keeps_stdout_open_for_sigterm_cleanup(self, tmp_path):
+ """
+ On an ``execution_timeout`` the timeout unwinds out of ``run_command``
and ``on_kill()``
+ sends SIGTERM only afterwards, so stdout has to still be open by then:
the child is started
+ with ``restore_signals=True``, so a TERM handler writing to a closed
pipe is killed by
+ SIGPIPE instead of finishing its cleanup.
+ """
+ marker = tmp_path / "cleanup-ran"
+ hook = SubprocessHook()
+
+ def raise_task_timeout(signum, frame):
+ raise _TaskTimeout()
+
+ previous_handler = signal.signal(signal.SIGALRM, raise_task_timeout)
+ try:
+ signal.setitimer(signal.ITIMER_REAL, 0.5)
+ with pytest.raises(_TaskTimeout):
+ hook.run_command(
+ command=[
+ "bash",
+ "-c",
+ f"trap 'echo cleaning up; touch {marker}; exit 1'
TERM; "
+ "echo started; while true; do sleep 0.1; done",
+ ]
+ )
+ finally:
+ signal.setitimer(signal.ITIMER_REAL, 0)
+ signal.signal(signal.SIGALRM, previous_handler)
+
+ hook.send_sigterm()
+ return_code = hook.sub_process.wait(timeout=10)
+ # run_command deliberately leaves this path's pipe open, so the test
owns closing it.
+ hook.sub_process.stdout.close()
+
+ assert marker.exists()
+ assert return_code == 1
+
def test_task_decode(self):
hook = SubprocessHook()
command = ["bash", "-c", 'printf "This will cause a coding error
\\xb1\\xa6\\x01\n"']
