Fix airflow-scheduler exiting with code 0 on exceptions

airflow/cli/commands/scheduler_command.py

@@ -19,7 +19,7 @@
 
 import logging
 from argparse import Namespace
-from contextlib import contextmanager
+from contextlib import ExitStack, contextmanager
 from multiprocessing import Process
 
 from airflow import settings
@@ -44,11 +44,18 @@ def _run_scheduler_job(args) -> None:
     ExecutorLoader.validate_database_executor_compatibility(job_runner.job.executor)
     InternalApiConfig.force_database_direct_access()
     enable_health_check = conf.getboolean("scheduler", "ENABLE_HEALTH_CHECK")
-    with _serve_logs(args.skip_serve_logs), _serve_health_check(enable_health_check):
+    with ExitStack() as stack:
+        stack.enter_context(_serve_logs(args.skip_serve_logs))
+        stack.enter_context(_serve_health_check(enable_health_check))
+
         try:
             run_job(job=job_runner.job, execute_callable=job_runner._execute)
         except Exception:
             log.exception("Exception when running scheduler job")
+            raise
+        finally:
+            # Ensure that the contexts are closed
+            stack.close()
 
 
 @cli_utils.action_cli

tests/cli/commands/test_scheduler_command.py

@@ -169,7 +169,8 @@ def test_run_job_exception_handling(
         mock_scheduler_job,
     ):
         args = self.parser.parse_args(["scheduler"])
-        scheduler_command.scheduler(args)
+        with pytest.raises(Exception, match="run_job failed"):
+            scheduler_command.scheduler(args)
 
         # Make sure that run_job is called, that the exception has been logged, and that the serve_logs
         # sub-process has been terminated