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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
