Skip to content

Keep queued asset events when a Dag is at max_active_runs #71998

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Vamsi-klu wants to merge 5 commits into
base: main
Choose a base branch
from
+535 −2
Open

Keep queued asset events when a Dag is at max_active_runs #71998

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6ca82cb
Keep queued asset events while a Dag is at max_active_runs
Vamsi-klu Aug 23, 2026
d4193a2
Address review feedback for max_active_runs asset queue tests
Vamsi-klu Aug 23, 2026
7eb0c44
Stop leftover asset events hiding behind a stale max_active_runs cache
Vamsi-klu Aug 23, 2026
4840348
Name the newsfragment after the pull request
Vamsi-klu Aug 23, 2026
64accdd
Name the newsfragment after the pull request
Vamsi-klu Aug 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions airflow-core/newsfragments/71998.bugfix.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Asset-triggered Dags no longer skip leftover queued events when a stale ``max_active_runs`` cache still says the Dag is at capacity.
1 change: 1 addition & 0 deletions airflow-core/src/airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2752,6 +2752,7 @@ def _create_dag_runs_asset_triggered(
)
stats.incr("asset.triggered_dagruns", tags=prune_dict({"team_name": team_name}))
dag_run.consumed_asset_events.extend(asset_events)
self._set_exceeds_max_active_runs(dag_model=dag_model, session=session)
self.log.info(
"Created asset-triggered DagRun for '%s': run_id=%s, consumed %d asset events",
dag.dag_id,
Expand Down
9 changes: 7 additions & 2 deletions airflow-core/src/airflow/models/dag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
Integer,
String,
Text,
and_,
case,
func,
inspect as sa_inspect,
Expand Down Expand Up @@ -776,9 +777,13 @@ def dag_ready(dag_id: str, cond: SerializedAssetBase, statuses: dict[UKey, bool]
cls.is_paused == expression.false(),
cls.is_stale == expression.false(),
cls.has_import_errors == expression.false(),
cls.exceeds_max_non_backfill == expression.false(),
# The cached flag is timetable-only. Asset leftovers stay selectable
# here; the live QUEUED+RUNNING count above already deferred Dags at cap.
or_(
cls.next_dagrun_create_after <= func.now(),
and_(
cls.exceeds_max_non_backfill == expression.false(),
cls.next_dagrun_create_after <= func.now(),
),
cls.dag_id.in_(asset_triggered_dag_ids),
),
)
Expand Down
Loading