From b19ac1f19277f048ef623cf5c4b837f66b049423 Mon Sep 17 00:00:00 2001 From: Xu Han Date: Mon, 6 Jul 2026 13:58:15 -0400 Subject: [PATCH 1/5] Add catch_exceptions option to EventScheduler.call_regular_interval _remove_unreferenced_triggers and _reap_stale_connection_tests are registered as periodic scheduler maintenance callbacks via EventScheduler.call_regular_interval, which has no exception handling around the callback. A single transient failure (e.g. a DB statement_timeout) propagates and crashes the whole SchedulerJob. Add an opt-in catch_exceptions flag (default False, preserving current behavior) that logs and swallows an exception from the periodic action instead of letting it propagate, so one bad cycle doesn't take down the scheduler; the next cycle is still scheduled either way. Enable it for the two maintenance callbacks above. --- .../src/airflow/jobs/scheduler_job_runner.py | 2 ++ .../src/airflow/utils/event_scheduler.py | 19 +++++++++-- .../tests/unit/utils/test_event_scheduler.py | 33 +++++++++++++++++++ 3 files changed, 52 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 7d5f7f58020c7..52b2bee3cdff4 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -1765,6 +1765,7 @@ def _run_scheduler_loop(self) -> None: timers.call_regular_interval( conf.getfloat("scheduler", "parsing_cleanup_interval"), self._remove_unreferenced_triggers, + catch_exceptions=True, ) if any(x.is_local for x in self.executors): @@ -1782,6 +1783,7 @@ def _run_scheduler_loop(self) -> None: timers.call_regular_interval( delay=conf.getfloat("connection_test", "reaper_interval", fallback=30.0), action=self._reap_stale_connection_tests, + catch_exceptions=True, ) idle_count = 0 diff --git a/airflow-core/src/airflow/utils/event_scheduler.py b/airflow-core/src/airflow/utils/event_scheduler.py index 88999ec69372f..2a32b22bdbb80 100644 --- a/airflow-core/src/airflow/utils/event_scheduler.py +++ b/airflow-core/src/airflow/utils/event_scheduler.py @@ -32,12 +32,27 @@ def call_regular_interval( action: Callable, arguments=(), kwargs=None, + catch_exceptions: bool = False, ): - """Call a function at (roughly) a given interval.""" + """ + Call a function at (roughly) a given interval. + + :param catch_exceptions: If True, an exception raised by ``action`` is logged + and swallowed instead of propagating, so a single bad cycle can't kill + whatever is driving this scheduler. The next cycle is still scheduled either + way. Defaults to False (propagate), preserving prior behavior for callers + that rely on the exception surfacing. + """ def repeat(*args, **kwargs): self.log.debug("Calling %s", action) - action(*args, **kwargs) + if catch_exceptions: + try: + action(*args, **kwargs) + except Exception: + self.log.exception("Exception raised in periodic action %s", action) + else: + action(*args, **kwargs) # This is not perfect. If we want a timer every 60s, but action # takes 10s to run, this will run it every 70s. # Good enough for now diff --git a/airflow-core/tests/unit/utils/test_event_scheduler.py b/airflow-core/tests/unit/utils/test_event_scheduler.py index 641d8dd0f909c..9b09edb974550 100644 --- a/airflow-core/tests/unit/utils/test_event_scheduler.py +++ b/airflow-core/tests/unit/utils/test_event_scheduler.py @@ -19,6 +19,8 @@ from unittest import mock +import pytest + from airflow.utils.event_scheduler import EventScheduler @@ -38,3 +40,34 @@ def test_call_regular_interval(self): assert len(timers.queue) == 2 somefunction.assert_called_once() assert timers.queue[0].time < timers.queue[1].time + + def test_call_regular_interval_propagates_exception_by_default(self): + """Without opting in, an action's exception still propagates (unchanged default behavior).""" + failing_action = mock.MagicMock(side_effect=RuntimeError("boom")) + + timers = EventScheduler() + timers.call_regular_interval(30, failing_action) + assert len(timers.queue) == 1 + + with pytest.raises(RuntimeError, match="boom"): + timers.queue[0].action() + + failing_action.assert_called_once() + # The next cycle was never scheduled because the exception propagated. + assert len(timers.queue) == 1 + + def test_call_regular_interval_catch_exceptions_swallows_action_exception(self): + """With catch_exceptions=True, a raising action is swallowed and the next cycle is still scheduled.""" + failing_action = mock.MagicMock(side_effect=RuntimeError("boom")) + + timers = EventScheduler() + timers.call_regular_interval(30, failing_action, catch_exceptions=True) + assert len(timers.queue) == 1 + + # Should not raise, even though the action does. + timers.queue[0].action() + + failing_action.assert_called_once() + # The next cycle was still scheduled despite the exception. + assert len(timers.queue) == 2 + assert timers.queue[0].time < timers.queue[1].time From 861f10d660e893856a79d9333071397fb0f26904 Mon Sep 17 00:00:00 2001 From: Xu Han Date: Mon, 6 Jul 2026 14:46:51 -0400 Subject: [PATCH 2/5] Rename call_regular_interval's catch_exceptions to non_fatal non_fatal reads better at call sites (e.g. non_fatal=True) than the mechanism-focused catch_exceptions, and matches existing "non-fatal" terminology already used elsewhere in the codebase for the same crash-avoidance concept (see dag_processing/importers/base.py). --- airflow-core/src/airflow/jobs/scheduler_job_runner.py | 4 ++-- airflow-core/src/airflow/utils/event_scheduler.py | 8 ++++---- airflow-core/tests/unit/utils/test_event_scheduler.py | 6 +++--- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 52b2bee3cdff4..874777ce1949c 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -1765,7 +1765,7 @@ def _run_scheduler_loop(self) -> None: timers.call_regular_interval( conf.getfloat("scheduler", "parsing_cleanup_interval"), self._remove_unreferenced_triggers, - catch_exceptions=True, + non_fatal=True, ) if any(x.is_local for x in self.executors): @@ -1783,7 +1783,7 @@ def _run_scheduler_loop(self) -> None: timers.call_regular_interval( delay=conf.getfloat("connection_test", "reaper_interval", fallback=30.0), action=self._reap_stale_connection_tests, - catch_exceptions=True, + non_fatal=True, ) idle_count = 0 diff --git a/airflow-core/src/airflow/utils/event_scheduler.py b/airflow-core/src/airflow/utils/event_scheduler.py index 2a32b22bdbb80..1461572b67424 100644 --- a/airflow-core/src/airflow/utils/event_scheduler.py +++ b/airflow-core/src/airflow/utils/event_scheduler.py @@ -32,13 +32,13 @@ def call_regular_interval( action: Callable, arguments=(), kwargs=None, - catch_exceptions: bool = False, + non_fatal: bool = False, ): """ Call a function at (roughly) a given interval. - :param catch_exceptions: If True, an exception raised by ``action`` is logged - and swallowed instead of propagating, so a single bad cycle can't kill + :param non_fatal: If True, an exception raised by ``action`` is logged and + swallowed instead of propagating, so a single bad cycle can't kill whatever is driving this scheduler. The next cycle is still scheduled either way. Defaults to False (propagate), preserving prior behavior for callers that rely on the exception surfacing. @@ -46,7 +46,7 @@ def call_regular_interval( def repeat(*args, **kwargs): self.log.debug("Calling %s", action) - if catch_exceptions: + if non_fatal: try: action(*args, **kwargs) except Exception: diff --git a/airflow-core/tests/unit/utils/test_event_scheduler.py b/airflow-core/tests/unit/utils/test_event_scheduler.py index 9b09edb974550..5a9e68b270ef0 100644 --- a/airflow-core/tests/unit/utils/test_event_scheduler.py +++ b/airflow-core/tests/unit/utils/test_event_scheduler.py @@ -56,12 +56,12 @@ def test_call_regular_interval_propagates_exception_by_default(self): # The next cycle was never scheduled because the exception propagated. assert len(timers.queue) == 1 - def test_call_regular_interval_catch_exceptions_swallows_action_exception(self): - """With catch_exceptions=True, a raising action is swallowed and the next cycle is still scheduled.""" + def test_call_regular_interval_non_fatal_swallows_action_exception(self): + """With non_fatal=True, a raising action is swallowed and the next cycle is still scheduled.""" failing_action = mock.MagicMock(side_effect=RuntimeError("boom")) timers = EventScheduler() - timers.call_regular_interval(30, failing_action, catch_exceptions=True) + timers.call_regular_interval(30, failing_action, non_fatal=True) assert len(timers.queue) == 1 # Should not raise, even though the action does. From f624db834fdccd12387601ce521d9fcc55497f6f Mon Sep 17 00:00:00 2001 From: Xu Han Date: Mon, 6 Jul 2026 16:48:41 -0400 Subject: [PATCH 3/5] Improve non_fatal log message with action name, exception, and a prompt to investigate Log "Failed to run periodic action due to . Please investigate if this is persistent." instead of a bare "Exception raised in periodic action ", making the log actionable on its own without needing to open the traceback to identify what failed. --- airflow-core/src/airflow/utils/event_scheduler.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/utils/event_scheduler.py b/airflow-core/src/airflow/utils/event_scheduler.py index 1461572b67424..695139d027789 100644 --- a/airflow-core/src/airflow/utils/event_scheduler.py +++ b/airflow-core/src/airflow/utils/event_scheduler.py @@ -49,8 +49,13 @@ def repeat(*args, **kwargs): if non_fatal: try: action(*args, **kwargs) - except Exception: - self.log.exception("Exception raised in periodic action %s", action) + except Exception as e: + self.log.exception( + "Failed to run periodic action %s due to %s. " + "Please investigate if this is persistent.", + getattr(action, "__name__", action), + e, + ) else: action(*args, **kwargs) # This is not perfect. If we want a timer every 60s, but action From ca30034a0a6c71f1ef9185d194ba5300b6b15ffb Mon Sep 17 00:00:00 2001 From: Xu Han Date: Mon, 6 Jul 2026 17:28:57 -0400 Subject: [PATCH 4/5] Match log message to house style, drop vague investigate prompt _update_dag_run_state_for_paused_dags already uses "Failed to due to %s" (scheduler_job_runner.py); match that instead of inventing new phrasing. Drop "please investigate if this is persistent" -- it's not actionable (nothing here can tell whether a single occurrence is persistent) and no other log line in this codebase phrases itself as a directive to the reader. Add "will retry on the next cycle" instead, since that's genuinely new information explaining why the exception is safe to swallow. --- airflow-core/src/airflow/utils/event_scheduler.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/utils/event_scheduler.py b/airflow-core/src/airflow/utils/event_scheduler.py index 695139d027789..298053ee3ee74 100644 --- a/airflow-core/src/airflow/utils/event_scheduler.py +++ b/airflow-core/src/airflow/utils/event_scheduler.py @@ -51,8 +51,7 @@ def repeat(*args, **kwargs): action(*args, **kwargs) except Exception as e: self.log.exception( - "Failed to run periodic action %s due to %s. " - "Please investigate if this is persistent.", + "Failed to run periodic action %s due to %s; will retry on the next cycle", getattr(action, "__name__", action), e, ) From 4739d6ea16a73c1ccf619398c96d7287b174006c Mon Sep 17 00:00:00 2001 From: Xu Han Date: Tue, 14 Jul 2026 11:24:36 -0400 Subject: [PATCH 5/5] Log non_fatal failures at warning level instead of error self.log.exception logs at ERROR with a full traceback. For a persistent (rather than transient) failure this repeats every cycle, creating steady-state log noise at ERROR severity. Log at WARNING instead, keeping exc_info=True so the traceback is still captured -- addresses review feedback without losing debuggability. --- airflow-core/src/airflow/utils/event_scheduler.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/utils/event_scheduler.py b/airflow-core/src/airflow/utils/event_scheduler.py index 298053ee3ee74..2d5dfa683ae80 100644 --- a/airflow-core/src/airflow/utils/event_scheduler.py +++ b/airflow-core/src/airflow/utils/event_scheduler.py @@ -50,10 +50,11 @@ def repeat(*args, **kwargs): try: action(*args, **kwargs) except Exception as e: - self.log.exception( + self.log.warning( "Failed to run periodic action %s due to %s; will retry on the next cycle", getattr(action, "__name__", action), e, + exc_info=True, ) else: action(*args, **kwargs)