Vamsi-klu opened a new pull request, #71996:
URL: https://github.com/apache/airflow/pull/71996
# Consume all mapped asset events on the next triggered run
closes: #54659
## What I did
Tests only. Mapped producer `@task(outlets=[asset]).expand(...)` succeeds,
then a scheduler tick. Assert all N events land on the consumer run and in
`triggering_asset_events`. Also pins empty expand, a failed map index,
leftovers on the next tick, AND/OR schedules, alias yields, catchup, and two
producers in the same second. No production change. No newsfragment.
## Why I did
Three mapped outlets were finishing together and the consumer only saw a
subset in `triggering_asset_events`. Consume-by-event-id is already on main.
Existing tests still insert ADRQ by hand, so they never covered this emit path.
These tests also drive one `dag_maker.run_ti` path.
## How I did
Emit: `register_asset_changes_in_db` after mapped SUCCESS, plus one
`dag_maker.run_ti(..., map_index=N)` case.
Consume: `SchedulerJobRunner._create_dagruns_for_dags`.
Context: `get_template_context` after loading `consumed_asset_events`.
AND waits for the second asset, then consumes the leftover queue. One tick
batches all visible events. That is the default, not a bug. Leftovers stay for
the next tick.
## What's the impact
None at runtime. If consume-by-id or the mapped emit path regresses, these
tests fail instead of silently dropping events.
## What's the testing
`airflow-core/tests/unit/jobs/test_scheduler_job.py`
- `test_mapped_asset_empty_expand_creates_no_run`
- `test_mapped_asset_events_consumed_together`
- `test_mapped_asset_run_ti_consumes_all_outlets`
- `test_mapped_asset_leftovers_consumed_on_next_tick`
- `test_mapped_asset_failed_index_not_queued`
- `test_mapped_asset_two_producers_same_second`
- `test_mapped_asset_alias_yields_consumed_together`
- `test_mapped_asset_and_condition_keeps_adrq`
- `test_mapped_asset_or_condition_consumes_fan_out`
- `test_mapped_asset_catchup_keeps_queued_events`
```
uv run --project airflow-core pytest
airflow-core/tests/unit/jobs/test_scheduler_job.py \
-k mapped_asset_ -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]
