Inconsistency between yielding Metadata and outlet_events extras for partitioned Assets
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
Airflow 3.2.1
### What happened and how to reproduce it?
I have a DAG that produces partitioned Assets. It contains a dynamicaly mapped task group of which one task has an Asset defined an outlet and emitts _AssetEvents_ for each expanded instance of the TaskGroup.
When I used `yield Metadata` the downstream DAG was triggered by each _AssetEvent_, so one run of the source DAG could result in e.g. three _DagRuns_ of the downstream DAG.
However, when I upgraded from Airflow 2.10.4 to Airflow 3.2.1, I also switched to attaching extra information to _AssetEvents_ with `outlet_events[ASSET].extra`.
The first issue I encountered was that _AssetEvents_ did not show up in the UI, when `outlet_events` wasn`t a keyword only argument in the producing task function. E.g.
```
def producing_function(input: str, outlet_events=None):
"""Example producing task"""
`outlet_events[ASSET].extra = {"section": "XY"}
```
Did not show AssetEvents. But it triggered the downstream DAG that then hang because the AssetEvent was not accessible.
Making _outlet_events_ a keyword argument like this:
```
def producing_function(input: str, *, outlet_events=None):
"""Example producing task"""
`outlet_events[ASSET].extra = {"section": "XY"}
```
produced visible _AssetEvents_. The downstream DAG gets triggered only once but shows all three _AssetEvents_ with the attached extra metadata from the three mapped taskgroup instances in the producing DAG.
When I use `yield Metadata(ASSET, {"section": "XY"})` in the producing task, each _TaskInstance_ triggers a DagRun of the downstream DAG with the extra metadata properly attached (which I expected to happen and how it was before).
The documentation indicates that `yield Metadata` and `outlet_events[ASSET].extra` are equivalent:
https://airflow.apache.org/docs/apache-airflow/3.2.2/authoring-and-scheduling/assets.html#attaching-extra-information-to-an-emitting-asset-event
For partitioned assets, they clearly are not.
### What you think should happen instead?
The difference in behavior between `yield Metadata(ASSET, {"section": "XY"})` and `outlet_events[ASSET].extra = {"section": "XY"}` - at least for partitioned DAGs - should be clearly documented and outlet_events should be properly produced also when outlet_events are not accessed as keyword only arguments.
### Operating System
Ubuntu 24.04
### Deployment
Docker-Compose
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
Not Applicable
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start with the linked Airflow assets documentation section on attaching extra information to emitting asset events, then reproduce the mapped partitioned-asset behavior using both Metadata and outlet_events[ASSET].extra. Compare event visibility, downstream DAG runs, and attached metadata; done means the behavior is consistent or the documented difference and keyword-only requirement are explicit.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 52/100