Vamsi-klu opened a new pull request, #71994:
URL: https://github.com/apache/airflow/pull/71994
# Consume every queued asset event from concurrent mapped outlets
closes: #54659
Independent of #71991 (same issue, different tests and branch).
## What I did
Tests only. Mapped producer `@task(outlets=[asset]).expand(...)` succeeds,
then one scheduler tick. Assert all N events land on the consumer run and in
`triggering_asset_events`. No production change. No newsfragment.
## Why I did
#54659 is three mapped outlets, one consumer run, `triggering_asset_events`
showing only one event. #70972 already switched ADRQ to `(target_dag_id,
asset_event_id)` and consume-by-id. Existing tests insert ADRQ by hand, so they
never covered this emit path. These tests are that coverage.
## How I did
Emit: `dag_maker.run_ti(..., map_index=N)` → `register_asset_changes_in_db`.
Consume: `SchedulerJobRunner._create_dagruns_for_dags`.
Context: `get_template_context` after loading `consumed_asset_events`.
One tick batches all visible events. That is the default, not a bug.
Leftovers stay for the next tick. SKIP LOCKED test was not changed.
## 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_outlet_asset_events_consumed_in_one_tick`
- `test_mapped_outlet_asset_events_consumed_across_staggered_ticks`
- `test_mapped_outlet_asset_events_same_timestamp_are_all_consumed`
- `test_mapped_outlet_asset_events_and_condition_waits_for_all_assets`
- `test_mapped_outlet_asset_alias_events_are_all_consumed`
Plus siblings `test_asset_events_out_of_order_are_both_consumed` and
`test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows`.
```
uv run --project airflow-core pytest
airflow-core/tests/unit/jobs/test_scheduler_job.py \
-k 'test_mapped_outlet or test_asset_events_out_of_order_are_both_consumed
or test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows' -q
```
7 passed.
---
##### 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]
