This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 5b04b42936f Avoid scheduler crash when periodic maintenance actions
fail (#69487)
5b04b42936f is described below
commit 5b04b42936f855d989fc1e3ad1c55c82c5800dad
Author: Xu Han <[email protected]>
AuthorDate: Fri Aug 7 12:46:08 2026 -0400
Avoid scheduler crash when periodic maintenance actions fail (#69487)
_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 ++
airflow-core/src/airflow/utils/event_scheduler.py | 24 ++++++++++++++--
.../tests/unit/utils/test_event_scheduler.py | 33 ++++++++++++++++++++++
3 files changed, 57 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 8b10948f7de..3f2c7c8a773 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -1842,6 +1842,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
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):
@@ -1859,6 +1860,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
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 88999ec6937..2d5dfa683ae 100644
--- a/airflow-core/src/airflow/utils/event_scheduler.py
+++ b/airflow-core/src/airflow/utils/event_scheduler.py
@@ -32,12 +32,32 @@ class EventScheduler(scheduler, LoggingMixin):
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.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)
# 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 641d8dd0f90..5a9e68b270e 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 __future__ import annotations
from unittest import mock
+import pytest
+
from airflow.utils.event_scheduler import EventScheduler
@@ -38,3 +40,34 @@ class TestEventScheduler:
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
