Vamsi-klu opened a new pull request, #71998:
URL: https://github.com/apache/airflow/pull/71998
# Keep queued asset events when a Dag is at max_active_runs
closes: #56050
This is leftover batch consume after the cap lifts, not one run per event.
## What I did
Reporter timeline: event 1 → run 1; events 2, 3, 4 while at
`max_active_runs=1` stay in ADRQ; after run 1 finishes, run 2 consumes
`{2,3,4}`; event 5 waits and becomes its own run.
`dags_needing_dagruns` no longer ANDs the cached `exceeds_max_non_backfill`
flag onto asset leftovers. Asset create sets that flag after `create_dagrun`.
Live QUEUED+RUNNING exclusion still defers a Dag at cap and does not delete
ADRQ.
## Why I did
Events 3 and 4 sat in the queue until a later event unstuck them. That used
to be cutoff-and-wipe. The wipe is already gone. The leftover hole is the stale
cache: asset create never refreshed `exceeds_max_non_backfill`, so a Dag with
queued events stayed hidden after the live cap lifted.
## How I did
The cached flag is timetable-only:
```
or_(
and_(
cls.exceeds_max_non_backfill == expression.false(),
cls.next_dagrun_create_after <= func.now(),
),
cls.dag_id.in_(asset_triggered_dag_ids),
)
```
After a successful asset `create_dagrun` and `consumed_asset_events.extend`,
call `_set_exceeds_max_active_runs`. Do not set it when no run is created. No
`session.commit()` in that path.
QUEUED counts toward the cap, not only RUNNING. A stuck MANUAL run counts
too. Catchup True/False does not change leftover consume. Partitioned same-key
follow-up: events 2/3/4 land on run 2, not on the already-created run 1.
## What's the impact
Asset leftovers stay visible once the live cap lifts, even if the cache
still says the Dag is full. The next run consumes the leftover batch. This does
not implement one run per event.
## What's the testing
`airflow-core/tests/unit/jobs/test_scheduler_job.py`
- `test_asset_events_queued_while_at_max_active_runs_are_all_consumed`
- `test_asset_events_wait_when_max_active_runs_is_two`
-
`test_already_consumed_adrq_after_max_active_runs_lift_does_not_create_empty_run`
- `test_partitioned_same_key_events_during_max_active_runs_land_on_next_run`
`airflow-core/tests/unit/models/test_dag.py`
- `test_dags_needing_dagruns_assets_stale_flag_does_not_hide_adrq`
- `test_dags_needing_dagruns_assets_retains_adrq_at_max_active_runs`
- `test_dags_needing_dagruns_assets_max_active_runs_two`
```
uv run --project airflow-core pytest \
airflow-core/tests/unit/jobs/test_scheduler_job.py \
airflow-core/tests/unit/models/test_dag.py \
-k 'max_active_runs or dags_needing_dagruns_assets' -q
```
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes (Grok 4.6)
Generated-by: Grok 4.6 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]
