You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Dev-iL (via GitHub)" <gi...@apache.org> on 2026/10/15 07:00:11 UTC
[PR] Close SubprocessHook stdout pipe after command completes [airflow]
Dev-iL opened a new pull request, #74780:
URL: https://github.com/apache/airflow/pull/74780
`SubprocessHook.run_command` never closed the stdout pipe of the `Popen` it creates. `Popen` does not close stdout after `wait()`, so the pipe stays open until the hook object is released, and it also stays open when reading the output raises. This change closes the pipe in a `finally` block, so it is closed on both the normal and the error path.
Why `try`/`finally` and not `with Popen(...)`: `Popen.__exit__` calls `wait()` for any exception other than `KeyboardInterrupt`. On an `execution_timeout`, `AirflowTaskTimeout` (a `BaseException`) is raised inside the read loop, and `on_kill()` runs only after the stack unwinds (`task-sdk/src/airflow/sdk/execution_time/task_runner.py`). With `with Popen(...)`, the unwind would block until the child exits, so `on_kill()` would never send SIGTERM to a long-running command.
## Evidence
- Descriptor leak: a 200-run loop that keeps each hook alive. Before the fix, +200 open file descriptors and 199 `ResourceWarning`s; after the fix, +0 and 0.
- Warnings under pytest: the hook and Bash operator, sensor and decorator test modules, run with `-W error::ResourceWarning`. Before the fix, 6 `PytestUnraisableExceptionWarning: ... unclosed file <_io.BufferedReader name=N>`; after the fix, 0.
- Full `providers/standard/tests/unit` suite on the fixed tree (run before the final test-only mock change): 1018 passed, 146 skipped, 0 failed. The hook test module was re-run after that change: 12 passed.
The leaked objects match the `unclosed file <_io.BufferedReader name=N>` warnings reported by `test_no_warnings` (`providers/standard/tests/unit/standard/decorators/test_python.py`), which runs with `filterwarnings("error")`. This PR does not claim to fix that test's flakiness: the CI failure was not reproduced locally.
<details><summary><h3>Reproducing the descriptor leak</h3></summary>
<p>
Save the script below as `dev/fd_leak_experiment.py`, then run it against the base and the fixed hook from the repository root:
```bash
git show upstream/main:providers/standard/src/airflow/providers/standard/hooks/subprocess.py > dev/subprocess_hook_base.py
breeze run python dev/fd_leak_experiment.py dev/subprocess_hook_base.py 200
breeze run python dev/fd_leak_experiment.py providers/standard/src/airflow/providers/standard/hooks/subprocess.py 200
rm dev/subprocess_hook_base.py dev/fd_leak_experiment.py
```
```python
from __future__ import annotations
import gc
import importlib.util
import os
import sys
import warnings
def load_hook_class(path: str) -> type:
spec = importlib.util.spec_from_file_location("hook_under_test", path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module.SubprocessHook
def open_fd_count() -> int:
return len(os.listdir("/proc/self/fd"))
def main() -> None:
hook_cls = load_hook_class(sys.argv[1])
runs = int(sys.argv[2])
baseline = open_fd_count()
kept: list = []
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
for _ in range(runs):
hook = hook_cls()
hook.run_command(["echo", "hi"])
kept.append(hook)
held = open_fd_count()
kept.clear()
released_by_refcount = open_fd_count()
gc.collect()
released_after_gc = open_fd_count()
resource_warnings = sum(1 for w in caught if issubclass(w.category, ResourceWarning))
print(
f"runs={runs} baseline_fds={baseline} "
f"fds_while_hooks_alive={held} (+{held - baseline}) "
f"after_dropping_refs={released_by_refcount} after_gc={released_after_gc} "
f"resource_warnings={resource_warnings}"
)
if __name__ == "__main__":
main()
```
</p>
</details>
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [x] Yes (please specify the tool below)
Generated-by: Claude Code following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Close SubprocessHook stdout pipe after command completes [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #74780:
URL: https://github.com/apache/airflow/pull/74780#discussion_r4306994587
##########
providers/standard/src/airflow/providers/standard/hooks/subprocess.py:
##########
@@ -93,12 +93,18 @@ def run_command(
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()
+ # Not `with Popen(...)`: its __exit__ blocks in wait(), so on an execution_timeout
+ # the task's on_kill() would never run to SIGTERM the child.
+ 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()
+ finally:
+ if self.sub_process.stdout is not None:
Review Comment:
Confirmed, and thank you for the reproduction! Added `test_run_command_keeps_stdout_open_for_sigterm_cleanup` for the path you found. It asserts the marker exists and the child exits `1` rather than `-SIGPIPE`, fails on the previous commit, and passes on this one.
Leaving this thread open for you to close since the fix isn't the one you asked for.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Close SubprocessHook stdout pipe after command completes [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 merged PR #74780:
URL: https://github.com/apache/airflow/pull/74780
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Close SubprocessHook stdout pipe after command completes [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on code in PR #74780:
URL: https://github.com/apache/airflow/pull/74780#discussion_r4306791560
##########
providers/standard/src/airflow/providers/standard/hooks/subprocess.py:
##########
@@ -93,12 +93,18 @@ def run_command(
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()
+ # Not `with Popen(...)`: its __exit__ blocks in wait(), so on an execution_timeout
+ # the task's on_kill() would never run to SIGTERM the child.
+ 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()
+ finally:
+ if self.sub_process.stdout is not None:
Review Comment:
**major** — Closing the pipe in `finally` also runs on the `execution_timeout` path, while the child is still alive and *before* `on_kill()` sends SIGTERM. The child was started with `restore_signals=True`, so SIGPIPE is at its default action: the next time it writes to stdout it is killed by SIGPIPE. A command whose SIGTERM handler writes anything (a very common `trap 'echo cleaning up; …' TERM` pattern) is now killed in the middle of its cleanup.
I reproduced it with the real hook, a `SIGALRM` that raises a `BaseException` inside `run_command` (as `AirflowTaskTimeout` does), then `hook.send_sigterm()` as `on_kill()` does:
```bash
trap 'echo cleaning up; touch $MARKER; exit 1' TERM; echo started; while true; do sleep 0.1; done
```
| hook | return code | cleanup ran |
|---|---|---|
| `main` | `1` | yes (3/3) |
| this PR | `-13` (SIGPIPE) | no (3/3) |
| this PR, close guarded by `poll() is not None` | `1` | yes (3/3) |
So the change defeats the guarantee the new code comment (and the PR body) gives for not using `with Popen(...)`: the child is no longer reliably terminated by `on_kill()`'s SIGTERM. Closing only once the child has exited keeps the fix for the normal and read-error paths (the pipe is still closed after a normal run, and the mocked read-error test still passes because `poll()` returns a mock), and leaves the timeout path as it is on `main`, where the hook is torn down shortly after anyway.
A regression test for this path would be good: run a `trap … TERM` command, interrupt `run_command` with a `BaseException`, call `send_sigterm()`, and assert the cleanup ran.
> **Flag any changed or added behaviour without a corresponding test, and flag tests that a reviewer cannot fail by reverting the PR's change.** The target is exactly 100% coverage of what the PR changes — no more, no less. Tests for pre-existing logic or standard-library/third-party functions are padding; missing tests for changed behaviour are gaps. Exception: deliberate behaviour or integration tests may cross those boundaries by design.
>
> — [.github/instructions/code-review.instructions.md](https://github.com/apache/airflow/blob/main/.github/instructions/code-review.instructions.md#testing-requirements)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Close SubprocessHook stdout pipe after command completes [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #74780:
URL: https://github.com/apache/airflow/pull/74780#discussion_r4306994587
##########
providers/standard/src/airflow/providers/standard/hooks/subprocess.py:
##########
@@ -93,12 +93,18 @@ def run_command(
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()
+ # Not `with Popen(...)`: its __exit__ blocks in wait(), so on an execution_timeout
+ # the task's on_kill() would never run to SIGTERM the child.
+ 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()
+ finally:
+ if self.sub_process.stdout is not None:
Review Comment:
Confirmed, and thank you for the reproduction! Added `test_run_command_keeps_stdout_open_for_sigterm_cleanup` for the path you found. It asserts the marker exists and the child exits `1` rather than `-SIGPIPE`, fails on the previous commit, and passes on this one.
The reason I did not use `poll() is not None` is the error path. On a real `readline` failure the child is still running, so `poll()` returns `None` and the pipe is never closed. It would also leave `test_run_command_closes_stdout_pipe_on_read_error` green only because `MagicMock.poll()` is truthy, as you noted yourself; a test that can't fail is worse than no test.
Leaving this thread open for you to close since the fix isn't the one you asked for.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org