(airflow) branch v3-1-test updated: [v3-1-test] Fix signal handling in triggerer job runner (#60190) (#60214)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/v3-1-test by this push:
     new 87e6b2eac03 [v3-1-test] Fix signal handling in triggerer job runner (#60190) (#60214)
87e6b2eac03 is described below

commit 87e6b2eac03201128038848a5b3f7fb3037909b1
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Wed Jan 7 14:51:36 2026 +0100

    [v3-1-test] Fix signal handling in triggerer job runner (#60190) (#60214)
    
    We were not registering our signal handler in the triggerer job. This
    meant our logging, to improve visibility into the shutdown process, was
    missing.
    (cherry picked from commit ee163cd6b70649d0b295c2f60abd515a26bb23fc)
    
    Co-authored-by: Jed Cunningham <66...@users.noreply.github.com>
---
 airflow-core/src/airflow/jobs/triggerer_job_runner.py | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 292c18f8799..c229dddc7fd 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -155,12 +155,13 @@ class TriggererJobRunner(BaseJobRunner, LoggingMixin):
             self.trigger_runner.stop = True
         else:
             self.log.warning("Forcing exit due to second exit signal %s", signum)
-
-            self.trigger_runner.kill(signal.SIGKILL)
+            if self.trigger_runner:
+                self.trigger_runner.kill(signal.SIGKILL)
             sys.exit(os.EX_SOFTWARE)
 
     def _execute(self) -> int | None:
         self.log.info("Starting the triggerer")
+        self.register_signals()
         try:
             # Kick off runner sub-process without DB access
             self.trigger_runner = TriggerRunnerSupervisor.start(
@@ -826,7 +827,6 @@ class TriggerRunner:
     failed_triggers: deque[tuple[int, BaseException | None]]
 
     # Should-we-stop flag
-    # TODO: set this in a sig-int handler
     stop: bool = False
 
     # TODO: connect this to the parent process
@@ -844,8 +844,14 @@ class TriggerRunner:
         self.failed_triggers = deque()
         self.job_id = None
 
+    def _handle_signal(self, signum, frame) -> None:
+        """Handle termination signals gracefully."""
+        self.stop = True
+
     def run(self):
-        """Sync entrypoint - just run a run in an async loop."""
+        """Sync entrypoint - just run arun in an async loop."""
+        signal.signal(signal.SIGINT, self._handle_signal)
+        signal.signal(signal.SIGTERM, self._handle_signal)
         asyncio.run(self.arun())
 
     async def arun(self):