diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 1718987cff683..5f2b699dcbca2 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -1747,6 +1747,7 @@ def _run_scheduler_loop(self) -> None: timers.call_regular_interval( conf.getfloat("scheduler", "parsing_cleanup_interval"), self._remove_unreferenced_triggers, + non_fatal=True, ) if any(x.is_local for x in self.executors): @@ -1764,6 +1765,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, + 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 88999ec69372f..298053ee3ee74 100644 --- a/airflow-core/src/airflow/utils/event_scheduler.py +++ b/airflow-core/src/airflow/utils/event_scheduler.py @@ -32,12 +32,31 @@ def call_regular_interval( action: Callable, arguments=(), kwargs=None, + non_fatal: bool = False, ): - """Call a function at (roughly) a given interval.""" + """ + Call a function at (roughly) a given interval. + + :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. + """ def repeat(*args, **kwargs): self.log.debug("Calling %s", action) - action(*args, **kwargs) + if non_fatal: + try: + action(*args, **kwargs) + except Exception as e: + self.log.exception( + "Failed to run periodic action %s due to %s; will retry on the next cycle", + getattr(action, "__name__", action), + e, + ) + 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..5a9e68b270ef0 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_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, non_fatal=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