Skip to content

Inconsistency between yielding Metadata and outlet_events extras for partitioned Assets #69117

Description

@ninsbl

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?

  • Yes I am willing to submit a PR!

Code of Conduct

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions