From f922c72af43e4762297d05de6eabb498ab5b951e Mon Sep 17 00:00:00 2001 From: ADITYA CHAUHAN <125276621+hooiv@users.noreply.github.com> Date: Fri, 24 Jul 2026 18:52:05 +0530 Subject: [PATCH 1/2] Fix supervisor hang by resetting SIGCHLD in CeleryExecutor --- .../providers/celery/executors/celery_executor_utils.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py index 39d52deb1dd13..f772eb14ab656 100644 --- a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py +++ b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py @@ -351,6 +351,12 @@ def _execute_in_fork(command_to_exec: CommandType, celery_task_id: str | None = if celery_task_id: args.external_executor_id = celery_task_id + # Reset SIGCHLD to default to prevent billiard's inherited handler from stealing + # the exit status of child processes spawned by the supervisor. + import signal + if hasattr(signal, "SIGCHLD"): + signal.signal(signal.SIGCHLD, signal.SIG_DFL) + setproctitle(f"airflow task supervisor: {command_to_exec}") log.debug("calling func '%s' with args %s", args.func.__name__, args) args.func(args) From e29d4d0f297678bfac9ba6b5759f918f7cab294a Mon Sep 17 00:00:00 2001 From: ADITYA CHAUHAN <125276621+hooiv@users.noreply.github.com> Date: Sat, 25 Jul 2026 09:35:37 +0530 Subject: [PATCH 2/2] Add test for SIGCHLD reset in celery executor --- .../celery/executors/test_celery_executor.py | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/providers/celery/tests/unit/celery/executors/test_celery_executor.py b/providers/celery/tests/unit/celery/executors/test_celery_executor.py index 28b0840590235..4df915418dd82 100644 --- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py +++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py @@ -1513,3 +1513,24 @@ def test_team_app_name_includes_team_name(self): team_conf = ExecutorConf(team_name="team_beta") celery_app = celery_executor_utils.create_celery_app(team_conf) assert "team_beta" in celery_app.main + + +class TestCeleryExecutorUtils: + @mock.patch("airflow.providers.celery.executors.celery_executor_utils.setproctitle") + @mock.patch("signal.signal") + def test_execute_in_fork_resets_sigchld(self, mock_signal, mock_setproctitle): + """ + Test that _execute_in_fork resets SIGCHLD to SIG_DFL to prevent + billiard from intercepting child process exit statuses. + """ + from airflow.providers.celery.executors import celery_executor_utils + import signal + + args = mock.MagicMock() + command_to_exec = ["airflow", "tasks", "run", "dummy"] + + celery_executor_utils._execute_in_fork(command_to_exec, celery_task_id="test_id", args=args) + + if hasattr(signal, "SIGCHLD"): + mock_signal.assert_called_once_with(signal.SIGCHLD, signal.SIG_DFL) + args.func.assert_called_once_with(args)