Close SubprocessHook stdout pipe after command completes

[Dev-iL]

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 ResourceWarnings; 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.

Reproducing the descriptor leak

Save the script below as dev/fd_leak_experiment.py, then run it against the base and the fixed hook from the repository root:

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

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()

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code following the guidelines

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.