ashb commented on code in PR #46677:
URL: https://github.com/apache/airflow/pull/46677#discussion_r1956800319
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -314,95 +116,222 @@ def on_kill(self):
Called when there is an external kill command (via the heartbeat
mechanism, for example).
"""
+ # TODO: signal instead.
self.trigger_runner.stop = True
- def _kill_listener(self):
- if self.listener:
- for h in self.listener.handlers:
- h.close()
- self.listener.stop()
-
def _exit_gracefully(self, signum, frame) -> None:
# The first time, try to exit nicely
- if not self.trigger_runner.stop:
+ if self.trigger_runner and not self.trigger_runner.stop:
self.log.info("Exiting gracefully upon receiving signal %s",
signum)
self.trigger_runner.stop = True
- self._kill_listener()
else:
self.log.warning("Forcing exit due to second exit signal %s",
signum)
+
+ self.trigger_runner.kill(signal.SIGKILL)
sys.exit(os.EX_SOFTWARE)
def _execute(self) -> int | None:
self.log.info("Starting the triggerer")
try:
- # set job_id so that it can be used in log file names
- self.trigger_runner.job_id = self.job.id
+ # Kick off runner sub-process without DB access
+ self.trigger_runner = TriggerRunnerMonitor.start(
+ job=self.job, capacity=self.capacity,
logger=structlog.get_logger(logger_name="triggerer2")
Review Comment:
Was testing stuff :D Needs tidying up clearly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
