This is an automated email from the ASF dual-hosted git repository.
dheerajturaga 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 4a011a8a7ab Fix airflow celery stop crashing on stale PID file (#72817)
4a011a8a7ab is described below
commit 4a011a8a7ab6fab9783ff4d27f328c97ad7aacba
Author: Y-C <[email protected]>
AuthorDate: Wed Sep 16 01:50:29 2026 +0800
Fix airflow celery stop crashing on stale PID file (#72817)
* Fix airflow celery stop crashing on stale PID file
When a Celery worker dies without cleaning up its PID file (SIGKILL,
OOM, host reboot), `airflow celery stop` reads the dead PID, fails with
an uncaught psutil.NoSuchProcess traceback, and never reaches the
PID-file removal. Every subsequent `stop` fails the same way until the
operator deletes the file by hand.
The worker start path already treats NoSuchProcess as "stale PID file,
clean it up" in check_if_pidfile_process_is_running; the stop path is
the only reader that did not. Only NoSuchProcess is handled on purpose:
AccessDenied means a live process the caller cannot signal, and its PID
file must not be removed.
* Apply batched suggestions from code review
Co-authored-by: rjgoyln <[email protected]>
---------
Co-authored-by: Eason09053360
<[email protected]>
Co-authored-by: Elad Kalif <[email protected]>
Co-authored-by: rjgoyln <[email protected]>
---
.../airflow/providers/celery/cli/celery_command.py | 9 +++--
.../tests/unit/celery/cli/test_celery_command.py | 38 +++++++++++++++-------
2 files changed, 34 insertions(+), 13 deletions(-)
diff --git
a/providers/celery/src/airflow/providers/celery/cli/celery_command.py
b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
index 830c5d42cc8..7a447534f25 100644
--- a/providers/celery/src/airflow/providers/celery/cli/celery_command.py
+++ b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
@@ -383,8 +383,13 @@ def stop_worker(args):
# Send SIGTERM
if pid:
- worker_process = psutil.Process(pid)
- worker_process.terminate()
+ try:
+ worker_process = psutil.Process(pid)
+ worker_process.terminate()
+ except psutil.NoSuchProcess:
+ log.warning(
+ "Worker process with PID %s is not running, PID file %s is
stale.", pid, pid_file_path
+ )
# Remove pid file
remove_existing_pidfile(pid_file_path)
diff --git a/providers/celery/tests/unit/celery/cli/test_celery_command.py
b/providers/celery/tests/unit/celery/cli/test_celery_command.py
index 9b9f4bcd7fd..b305993f657 100644
--- a/providers/celery/tests/unit/celery/cli/test_celery_command.py
+++ b/providers/celery/tests/unit/celery/cli/test_celery_command.py
@@ -27,6 +27,7 @@ from io import StringIO
from unittest import mock
from unittest.mock import MagicMock, patch
+import psutil
import pytest
from airflow.cli import cli_parser
@@ -73,23 +74,38 @@ class TestCeleryStopCommand:
importlib.reload(cli_parser)
cls.parser = cli_parser.get_parser()
- @mock.patch("airflow.providers.celery.cli.celery_command.setup_locations")
- @mock.patch("airflow.providers.celery.cli.celery_command.psutil.Process")
- def test_if_right_pid_is_read(self, mock_process, mock_setup_locations,
tmp_path):
+ @pytest.mark.parametrize(
+ ("process_side_effect", "terminate_side_effect",
"expected_terminate_calls"),
+ [
+ pytest.param(None, None, 1, id="worker-running"),
+ pytest.param(psutil.NoSuchProcess(123), None, 0,
id="stale-pid-file"),
+ pytest.param(None, psutil.NoSuchProcess(123), 1,
id="worker-exits-before-terminate"),
+ ],
+ )
+ @mock.patch("airflow.providers.celery.cli.celery_command.setup_locations",
autospec=True)
+ @mock.patch("airflow.providers.celery.cli.celery_command.psutil.Process",
autospec=True)
+ def test_stop_worker_handles_missing_process(
+ self,
+ mock_process,
+ mock_setup_locations,
+ process_side_effect,
+ terminate_side_effect,
+ expected_terminate_calls,
+ tmp_path,
+ ):
args = self.parser.parse_args(["celery", "stop"])
- pid = "123"
+ pid = 123
path = tmp_path / "testfile"
- # Create pid file
- path.write_text(pid)
- # Setup mock
+ path.write_text(str(pid))
mock_setup_locations.return_value = (os.fspath(path), None, None, None)
+ mock_process.side_effect = process_side_effect
+ mock_process.return_value.terminate.side_effect = terminate_side_effect
- # Calling stop_worker should delete the temporary pid file
celery_command.stop_worker(args)
- # Check if works as expected
+
assert not path.exists()
- mock_process.assert_called_once_with(int(pid))
- mock_process.return_value.terminate.assert_called_once_with()
+ mock_process.assert_called_once_with(pid)
+ assert mock_process.return_value.terminate.call_args_list ==
[mock.call()] * expected_terminate_calls
@mock.patch("airflow.providers.celery.cli.celery_command.read_pid_from_pidfile")
@mock.patch("airflow.providers.celery.executors.celery_executor.app")
