muditagrawal2007 opened a new pull request, #75147: URL: https://github.com/apache/airflow/pull/75147
Fixes a sharp edge in `SchedulerJobRunner._remove_unreferenced_triggers` left behind when the triggerer's cleanup was batched in #70961 — the scheduler-side counterpart was missed, and the prior attempt (#68241, dkranchii) was closed as a stale draft. ## What `SchedulerJobRunner._remove_unreferenced_triggers` ran as a synchronous `call_regular_interval` callback every `parsing_cleanup_interval` (default 60s) and issued a single unbounded `DELETE FROM trigger` with `synchronize_session="fetch"`. On installs with a large trigger table that holds row locks for the entire transaction — blocking the triggerer's own writers (`Trigger.clean_unused`, `submit_event`, `submit_failure`) and stalling the scheduler main loop. `synchronize_session="fetch"` also forces a `SELECT id` over every match before the `DELETE`, materialising every row in Python memory. Reuses the proven `Trigger._delete_unused_batch` helper from `models/trigger.py` (the same one the triggerer landed on in #70961): `select id LIMIT, with_row_locks(skip_locked=True), delete by id, commit`. New config option `[scheduler] unreferenced_triggers_cleanup_batch_size` (default 500) keeps the scheduler and triggerer tunable independently — `0`/`-1` raises `ValueError` instead of silently running the unbounded form. ## Why reuse, not re-roll - Single source of truth: any future tightening of the lock policy or row-skip semantics edits one method. - The triggerer's helper already uses `with_row_locks(skip_locked=True)` — needed for concurrent cleanup between scheduler and triggerer on the same trigger rows. PR #68241's inlined loop omitted this; cleaning it up here. - Reproduces `Trigger.clean_unused`'s validation contract (`batch_size > 0`), so both cleanup entry points reject the bad value identically. ## Cadence One batch per call, matching `Trigger.clean_unused`. Earlier PR #68241 drained the entire backlog per tick (one tick held locks longer); this draft chooses the triggerer's pattern instead, so per-tick work is bounded regardless of backlog size. The remainder drains over subsequent `parsing_cleanup_interval` ticks — a 100 000-row backlog at default values finishes in roughly 100 000 / 500 ≈ 200 ticks (≈ 3.3h). Dead rows don't block queue heads, so the slower drain is acceptable and the lower peak load is what unblocks the triggerer's writers. ## Tests New tests in `tests/unit/jobs/test_scheduler_job.py`: - `test_remove_unreferenced_triggers_drains_in_batches_per_call` — 5 triggers, `batch_size=2` → exactly the triggerer's cadence pattern of (3, 1, 0) remaining across three calls. Mirrors `test_clean_unused_deletes_one_batch_per_call` in `tests/unit/models/test_trigger.py`. - `test_remove_unreferenced_triggers_rejects_non_positive_batch_size` — parametrized over `0` and `-1`. Asserts the message contract and that the trigger row survives (no DELETE was issued). - `test_remove_unreferenced_triggers_preserves_referenced_triggers` — TI tie → referenced survives, unreferenced is deleted. Asset-watcher and Callback preservation contracts already pin down in the triggerer's own tests. Existing `test_delete_unreferenced_triggers` (4 parametrized cases) and `test_delete_unreferenced_triggers_with_null_trigger_id_ti` pass unchanged at default `batch_size=500`. Full module suite (558 passed, 12 skipped for SQLite-only constraints not related to this change). ## Diff 3 files, +122 / -13: ``` airflow-core/src/airflow/jobs/scheduler_job_runner.py | 29 +++-- airflow-core/src/airflow/config_templates/config.yml | 12 +++ airflow-core/tests/unit/jobs/test_scheduler_job.py | 94 ++++++++++++ ``` ## Validation - `prek run --files …` (no-fail docs/ruff/mypy/codespell/no-AirflowException/Mapped[X|None]) — passed. - `mypy airflow-core/src/airflow/jobs/scheduler_job_runner.py` — `Success: no issues found`. - `pytest airflow-core/tests/unit/jobs/test_scheduler_job.py` — 558 passed, 12 skipped. - `pytest airflow-core/tests/unit/models/test_trigger.py` — 66 passed (no model change, smoke-tested). - Codespell + ruff-format on changed files clean. - The CI-only `Check the default configuration` (Breeze) hook is the one I couldn't run locally; it spins up a Docker container to verify the rendered airflow.cfg. I'll watch the CI logs if/when the PR reaches the build docs job. ## No newsfragment Per AGENTS.md / contributing-docs/18: newsfragments in `airflow-core/` are reserved for `significant` changes that users must read before upgrading (behaviour change, removal, new default, migration steps). This is a perf-bug fix with no user-facing behaviour shift: a new optional config knob with a sensible default is exposed, and the visible cleanup behaviour is "fewer cleanup-related stalls on the trigger table" — not "different cleanup counts over time." A maintainer can ask for one at review if they disagree. closes: #75141 ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Minimax-M3 Generated-by: Minimax-M3 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
