This is an automated email from the ASF dual-hosted git repository.
henry3260 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 84128892672 Stop airflow standalone leaking components when one fails
to start (#72464)
84128892672 is described below
commit 841288926725a61919fb96b92a648929081ef0e4
Author: Y-C <[email protected]>
AuthorDate: Sun Sep 6 01:58:37 2026 +0800
Stop airflow standalone leaking components when one fails to start (#72464)
* Stop airflow standalone leaking components when one fails to start
SubCommand.process is only assigned once the thread reaches Popen, so a
component whose executable could not be spawned leaves the attribute absent
rather than empty. Shutdown then raises out of the loop that terminates each
component in turn, and every sibling after the failed one keeps running
against the metadata database after standalone itself has exited.
* Update airflow-core/tests/unit/cli/commands/test_standalone_command.py
---------
Co-authored-by: Eason09053360
<[email protected]>
Co-authored-by: Henry Chen <[email protected]>
---
airflow-core/src/airflow/cli/commands/standalone_command.py | 4 +++-
.../tests/unit/cli/commands/test_standalone_command.py | 13 +++++++++++++
2 files changed, 16 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/cli/commands/standalone_command.py
b/airflow-core/src/airflow/cli/commands/standalone_command.py
index 2e94637e763..97b327b426b 100644
--- a/airflow-core/src/airflow/cli/commands/standalone_command.py
+++ b/airflow-core/src/airflow/cli/commands/standalone_command.py
@@ -291,6 +291,7 @@ class SubCommand(threading.Thread):
self.name = name
self.command = command
self.env = env
+ self.process: subprocess.Popen[bytes] | None = None
def run(self):
"""Run the actual process and captures it output to a queue."""
@@ -305,7 +306,8 @@ class SubCommand(threading.Thread):
def stop(self):
"""Call to stop this process (and thus this thread)."""
- self.process.terminate()
+ if self.process is not None:
+ self.process.terminate()
# Alias for use in the CLI parser
diff --git a/airflow-core/tests/unit/cli/commands/test_standalone_command.py
b/airflow-core/tests/unit/cli/commands/test_standalone_command.py
index 993f51ca2d7..b2e7c2496fa 100644
--- a/airflow-core/tests/unit/cli/commands/test_standalone_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_standalone_command.py
@@ -18,6 +18,7 @@
from __future__ import annotations
import os
+import subprocess
from collections import deque
from importlib import reload
from unittest import mock
@@ -304,6 +305,18 @@ class TestStandaloneCommand:
fake_process.terminate.assert_called_once()
+ def
test_subcommand_stop_does_not_block_siblings_when_a_process_never_started(self):
+ """Shutdown stops each component in turn, so one that never spawned
must not abort the loop."""
+ never_started = SubCommand(mock.Mock(spec=StandaloneCommand),
"scheduler", ["scheduler"], {})
+ running = SubCommand(mock.Mock(spec=StandaloneCommand), "triggerer",
["triggerer"], {})
+ running.process = mock.Mock(spec=subprocess.Popen)
+
+ assert never_started.process is None
+ for command in (never_started, running):
+ command.stop()
+
+ running.process.terminate.assert_called_once()
+
@mock.patch("airflow.cli.commands.standalone_command.ExecutorLoader.import_default_executor_cls")
@mock.patch("airflow.cli.commands.standalone_command.conf.get")
@mock.patch.dict(os.environ, {}, clear=True)
