This is an automated email from the ASF dual-hosted git repository.
potiuk 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 2e7207f32ac Make blocked-thread warning threshold configurable (#65009)
2e7207f32ac is described below
commit 2e7207f32ac7f9721eb511f788416757882ac9f7
Author: Ulada Zakharava <[email protected]>
AuthorDate: Mon Apr 27 23:20:58 2026 +0000
Make blocked-thread warning threshold configurable (#65009)
---
.../src/airflow/config_templates/config.yml | 9 ++++
.../src/airflow/jobs/triggerer_job_runner.py | 11 +++--
airflow-core/tests/unit/jobs/test_triggerer_job.py | 51 ++++++++++++++++++++++
3 files changed, 68 insertions(+), 3 deletions(-)
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index b36ef5b4dc3..c6dc3e97f29 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -2685,6 +2685,15 @@ triggerer:
type: float
example: ~
default: "30"
+ blocked_main_thread_warning_threshold:
+ description: |
+ Threshold in seconds for logging a warning when the Triggerer's async
thread appears blocked.
+ Increase this value if your deployment can tolerate longer delays in
the Triggerer's async
+ loop before logging a warning.
+ version_added: 3.2.0
+ type: float
+ example: ~
+ default: "0.2"
max_trigger_to_select_per_loop:
description: |
Maximum number of triggers to select per loop. Set this notably lower
than ``[triggerer] capacity``
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 2d0fc2ae224..e1c6f1ccb1c 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -984,6 +984,9 @@ class TriggerRunner:
self.failed_triggers = deque()
self.job_id = None
self._stop_event = None
+ self.blocked_main_thread_warning_threshold = conf.getfloat(
+ "triggerer", "blocked_main_thread_warning_threshold"
+ )
def _handle_signal(self, signum, frame) -> None:
"""Handle termination signals gracefully."""
@@ -1303,12 +1306,14 @@ class TriggerRunner:
# We allow a generous amount of buffer room for now, since it might
# be a busy event loop.
time_elapsed = time.monotonic() - last_run
- if time_elapsed > 0.2:
+ if time_elapsed > self.blocked_main_thread_warning_threshold:
await self.log.ainfo(
"Triggerer's async thread was blocked for %.2f seconds, "
- "likely by a badly-written trigger. Set
PYTHONASYNCIODEBUG=1 "
- "to get more information on overrunning coroutines.",
+ "exceeding the configured warning threshold of %.2f
seconds. "
+ "This is likely caused by a badly-written trigger. "
+ "Set PYTHONASYNCIODEBUG=1 to get more information on
overrunning coroutines.",
time_elapsed,
+ self.blocked_main_thread_warning_threshold,
)
Stats.incr("triggers.blocked_main_thread")
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index ef5b5a02c5c..8355fa13262 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -74,6 +74,7 @@ from airflow.triggers.testing import FailureTrigger,
SuccessTrigger
from airflow.utils.state import State, TaskInstanceState
from airflow.utils.types import DagRunType
+from tests_common.test_utils.config import conf_vars
from tests_common.test_utils.db import (
clear_db_connections,
clear_db_dag_bundles,
@@ -407,6 +408,56 @@ def test_trigger_logger_fd_closed_when_removed(session):
class TestTriggerRunner:
+ def test_blocked_main_thread_warning_threshold_decode(self) -> None:
+ with conf_vars({("triggerer",
"blocked_main_thread_warning_threshold"): "0.5"}):
+ trigger_runner = TriggerRunner()
+
+ assert trigger_runner.blocked_main_thread_warning_threshold == 0.5
+
+ @pytest.mark.asyncio
+ async def
test_block_watchdog_does_not_log_when_threshold_is_not_exceeded(self) -> None:
+ with conf_vars({("triggerer",
"blocked_main_thread_warning_threshold"): "0.5"}):
+ trigger_runner = TriggerRunner()
+
+ trigger_runner.log = AsyncMock()
+
+ async def fake_sleep(_):
+ trigger_runner.stop = True
+
+ with (
+ patch("airflow.jobs.triggerer_job_runner.asyncio.sleep",
side_effect=fake_sleep),
+ patch("airflow.jobs.triggerer_job_runner.time.monotonic",
side_effect=[1.0, 1.4]),
+ patch("airflow.jobs.triggerer_job_runner.Stats.incr") as
mock_stats_incr,
+ ):
+ await trigger_runner.block_watchdog()
+
+ trigger_runner.log.ainfo.assert_not_called()
+ mock_stats_incr.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_block_watchdog_logs_when_threshold_is_exceeded(self) ->
None:
+ with conf_vars({("triggerer",
"blocked_main_thread_warning_threshold"): "0.5"}):
+ trigger_runner = TriggerRunner()
+
+ trigger_runner.log = AsyncMock()
+
+ async def fake_sleep(_):
+ trigger_runner.stop = True
+
+ with (
+ patch("airflow.jobs.triggerer_job_runner.asyncio.sleep",
side_effect=fake_sleep),
+ patch("airflow.jobs.triggerer_job_runner.time.monotonic",
side_effect=[1.0, 1.6]),
+ patch("airflow.jobs.triggerer_job_runner.Stats.incr") as
mock_stats_incr,
+ ):
+ await trigger_runner.block_watchdog()
+
+ trigger_runner.log.ainfo.assert_awaited_once()
+ log_message, elapsed, threshold =
trigger_runner.log.ainfo.await_args.args
+ assert "configured warning threshold" in log_message
+ assert elapsed == pytest.approx(0.6)
+ assert threshold == 0.5
+ mock_stats_incr.assert_called_once_with("triggers.blocked_main_thread")
+
def test_run_inline_trigger_canceled(self, session) -> None:
trigger_runner = TriggerRunner()
trigger_runner.triggers = {
