Outlet Event Extra Data is Empty in Task Instance Success Listener

[dygger]

Apache Airflow version

3.0.2

If "Other Airflow 2 version" selected, which one?

No response

What happened?

When accessing outlet event extra data in the on_task_instance_success listener hook, the extra field returns an empty dict even though the task function explicitly sets non-empty extra data.

What you think should happen instead?

The extra field in outlet events should contain the data set by the task and be accessible in the listener hook.

How to reproduce

DAGs

with DAG(
    dag_id="asset-producer",
    dagrun_timeout=timedelta(minutes=10),
    start_date=datetime(2021, 1, 1),
    schedule=None,
    catchup=False,
    is_paused_upon_creation=False,
    tags=["test"],
):

    @task(outlets=[asset])
    def produce(
        *,
        params: dict | None = None,
        outlet_events=None,
    ):
        outlet_events[asset].extra = {"name1": "value1", "nested_obj": {"name2": "value2"}}
        logger.info(f"Outlet events: {outlet_events}")

    produce()


with DAG(
    dag_id="asset-consumer",
    start_date=datetime(2021, 1, 1),
    schedule=[asset],
    catchup=False,
    is_paused_upon_creation=False,
    tags=["test"],
):

    @task(inlets=[asset])
    def consume(*, triggering_asset_events=None):
        for k, v in triggering_asset_events.items():
            logger.info(f"[Consumer] triggering_asset_event: {k}: {v}")

    consume()

Listener

@hookimpl
def on_task_instance_success(
    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
):
    logger.info(f"on_task_instance_success: {task_instance.task_id}")
    if isinstance(task_instance, RuntimeTaskInstance):
        context = task_instance.get_template_context()
        outlets = context["outlets"]
        logger.info(f"on_task_instance_success outlets: {outlets}")
        outlet_events = context["outlet_events"]
        for outlet in outlets:
            event = outlet_events[outlet]
            logger.info(f"on_task_instance_success outlet event: {event}")

Logs

Producer (event extra is empty in the listener)

[2025-06-30, 16:56:00] INFO - Outlet events: OutletEventAccessors(_dict={AssetUniqueKey(name='test-asset', uri='test-asset'): OutletEventAccessor(key=AssetUniqueKey(name='test-asset', uri='test-asset'), extra={'name1': 'value1', 'nested_obj': {'name2': 'value2'}}, asset_alias_events=[])}): source="unusual_prefix_e7352dd1f756ead6f66dfc[4](http://localhost:8080/dags/asset-producer/runs/manual__2025-06-30T13:55:55.913030+00:00/tasks/produce?try_number=1#4)89eeb1b1519f4e074_outlet_test_dag"
[2025-06-30, 16:56:00] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.decorators.python._PythonDecoratedOperator"
[2025-06-30, 16:56:00] INFO - on_task_instance_success: produce: source="pipeline_run_listener"
[2025-06-30, 16:56:00] INFO - on_task_instance_success outlets: [Asset(name='test-asset', uri='test-asset', group='asset', extra={}, watchers=[])]: source="pipeline_run_listener"
[202[5](http://localhost:8080/dags/asset-producer/runs/manual__2025-06-30T13:55:55.913030+00:00/tasks/produce?try_number=1#5)-06-30, 16:56:00] INFO - on_task_instance_success outlet event: OutletEventAccessor(key=AssetUniqueKey(name='test-asset', uri='test-asset'), extra={}, asset_alias_events=[]): source="pipeline_run_listener"

Consumer (event extra is correct)

[2025-06-30, 16:56:03] INFO - [Consumer] triggering_asset_event: Asset(name='test-asset', uri='test-asset', group='asset', extra={}, watchers=[]): [AssetEventDagRunReferenceResult(asset=AssetReferenceAssetEventDagRun(name='test-asset', uri='test-asset', extra={}), extra={'name1': 'value1', 'nested_obj': {'name2': 'value2'}}, source_task_id='produce', source_dag_id='asset-producer', source_run_id='manual__2025-06-30T13:55:55.913030+00:00', source_map_index=-1, source_aliases=[], timestamp=datetime.datetime(2025, 6, 30, 13, 56, 0, 859276, tzinfo=TzInfo(UTC)))]: source="unusual_prefix_e7352dd1f756ead6f66dfc[4](http://localhost:8080/dags/asset-consumer/runs/asset_triggered__2025-06-30T13:56:00.882558+00:00_orhaCyKJ/tasks/consume?try_number=1#4)89eeb1b1519f4e074_outlet_test_dag"
[2025-06-30, 16:56:03] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.decorators.python._PythonDecoratedOperator"

Operating System

Ubuntu 24.04.2 LTS

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[dygger]

Any chance to review/triage this? Or maybe there is an another way to retrieve the missing extra in on_task_instance_success listener?

[sjyangkevin]

Hi @dygger , I tried to reproduce the error, but I am a bit confused by the DAG code. Is the listener defined in the same file as your producer and consumer DAG? Would you mind provide the complete DAG definition so I can use it to reproduce the error?

[dygger]

@sjyangkevin, thank you for looking into this issue.

The listener is defined in a separate file test_listener.py and is registered as part of Airflow plugin (see test_plugin.py) as described at Listener Registration section. Both files should be placed in $AIRFLOW_HOME/plugins/ folder.

plugins/test_plugin.py

import test_listener
from airflow.plugins_manager import AirflowPlugin


class TestPlugin(AirflowPlugin):
    name = "TestPlugin"
    listeners = [test_listener]

plugins/test_listener.py

from __future__ import annotations

import logging

from airflow.listeners import hookimpl
from airflow.models.taskinstance import TaskInstance
from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
from airflow.utils.state import TaskInstanceState

logger = logging.getLogger(__name__)


@hookimpl
def on_task_instance_success(
    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
):
    logger.info(f"on_task_instance_success: {task_instance.task_id}")
    if isinstance(task_instance, RuntimeTaskInstance):
        context = task_instance.get_template_context()
        outlets = context["outlets"]
        logger.info(f"on_task_instance_success outlets: {outlets}")
        outlet_events = context["outlet_events"]
        for outlet in outlets:
            event = outlet_events[outlet]
            logger.info(f"on_task_instance_success outlet event: {event}")

dags/outlet_test_dag.py

from datetime import datetime

from airflow.sdk import DAG, Asset, task

asset = Asset("test-asset")

with DAG(
    dag_id="asset-producer",
    start_date=datetime(2021, 1, 1),
    schedule=None,
    catchup=False,
    is_paused_upon_creation=False,
    tags=["test"],
):

    @task(outlets=[asset])
    def produce(
        *,
        outlet_events=None,
    ):
        outlet_events[asset].extra = {"name1": "value1", "nested_obj": {"name2": "value2"}}
        print(f"Outlet events: {outlet_events}")

    produce()


with DAG(
    dag_id="asset-consumer",
    start_date=datetime(2021, 1, 1),
    schedule=[asset],
    catchup=False,
    is_paused_upon_creation=False,
    tags=["test"],
):

    @task(inlets=[asset])
    def consume(*, triggering_asset_events=None):
        for k, v in triggering_asset_events.items():
            print(f"[Consumer] triggering_asset_event: {k}: {v}")

    consume()

[sjyangkevin]

looking into it

[sjyangkevin]

After observing the metadata store, I think the issue is because the following:

First, the asset itself doesn't contain any extra.

When you do outlet_events[asset].extra = {"name1": "value1", "nested_obj": {"name2": "value2"}}, the extra is attached to the asset event.

The extra is propagated into the consumer DAG, and can be accessed through either consumed_asset_events or triggering_asset_events.

Looking at the code below that populate the context for task instance, it looks like both InletEventsAccessors and OutletEventAccessors are querying for the Asset model. However, the extra is attached to the asset event instead of the asset. So, it is why when accessing the outlet_events in the listener, the extra is empty.

airflow/airflow-core/src/airflow/models/taskinstance.py

Lines 1922 to 1925 in 0cd0ff0

	context.update(
	{
	"outlet_events": OutletEventAccessors(),
	"inlet_events": InletEventsAccessors(task.inlets),

airflow/airflow-core/src/airflow/utils/context.py

Lines 126 to 148 in 0cd0ff0

	class OutletEventAccessors(OutletEventAccessorsSDK):
	"""
	Lazy mapping of outlet asset event accessors.
	:meta private:
	"""
	@staticmethod
	def _get_asset_from_db(name: str | None = None, uri: str | None = None) -> Asset:
	if name:
	with create_session() as session:
	asset = session.scalar(
	select(AssetModel).where(AssetModel.name == name, AssetModel.active.has())
	)
	elif uri:
	with create_session() as session:
	asset = session.scalar(
	select(AssetModel).where(AssetModel.uri == uri, AssetModel.active.has())
	)
	else:
	raise ValueError("Either name or uri must be provided")
	return asset.to_public()

airflow/task-sdk/src/airflow/sdk/execution_time/context.py

Lines 503 to 512 in 0cd0ff0

	if isinstance(inlet, Asset):
	self._assets[AssetUniqueKey.from_asset(inlet)] = inlet
	elif isinstance(inlet, AssetAlias):
	self._asset_aliases[AssetAliasUniqueKey.from_asset_alias(inlet)] = inlet
	elif isinstance(inlet, AssetNameRef):
	asset = OutletEventAccessors._get_asset_from_db(name=inlet.name)
	self._assets[AssetUniqueKey.from_asset(asset)] = asset
	elif isinstance(inlet, AssetUriRef):
	asset = OutletEventAccessors._get_asset_from_db(uri=inlet.uri)
	self._assets[AssetUniqueKey.from_asset(asset)] = asset

It would be great if @uranusjr and @Lee-W could kindly have a look at this, and correct me if anything is wrong here. I am not sure if it is by-design, or we should adjust these two accessors to query for the asset event associated with this asset trigger. Thanks.

[uranusjr]

Asset extra and asset event extra are two different concepts. I don’t think you’re looking at the right things @sjyangkevin.

The original issue reported seems to be legistimate (assuming the description is accurate; I did not try to reproduce); both the downstream dag and on_task_instance_success query the asset event, and they should get the same event extra.

[sjyangkevin]

Asset extra and asset event extra are two different concepts. I don’t think you’re looking at the right things @sjyangkevin.

The original issue reported seems to be legistimate (assuming the description is accurate; I did not try to reproduce); both the downstream dag and on_task_instance_success query the asset event, and they should get the same event extra.

Thanks for pointing this out. I will try to understand it better and have a deeper look into it.

[uranusjr]

I took a quick look at the code. In task_runner.py, the task runner does not commit the asset changes to the database (by sending a request through API) before calling the callbacks and listeners (not a bad implementation; those functions may further modify the task instance so we want to wait until they are finished), but when you call get_template_context in the listener you get a brand new context object that does not contain changes made during task execution.

If my understanding is correct, the easiest fix is probably to add a cache to get_template_context so all calls inside a task worker always return the same context object instead of building new ones.

[sjyangkevin]

Thanks for the insightful feedback. I am trying to understand the mechanism better.

the task runner does not commit the asset changes to the database before calling the callbacks and listeners.

when you call get_template_context in the listener you get a brand new context object that does not contain changes made during task execution

all calls inside a task worker always return the same context object instead of building new ones

If I understanding correctly, the context object is updated during the task execution. So, when attaching the extra to the asset event, the context object is updated with the changes. However, in the listener, the call to the get_template_context method construct a new object and returns it. The object with the changes then lost, and we will never be able to get it back? So, adding a cache here, we can ensure all the calls to get_template_context is operating on the same context object, so the changes are persist.

[uranusjr]

The task runner keeps track of the modified object and uses it to submit the changes. It is the context retrieved in the listener callback that does not reflect the modified object, and changes to the context created inside the listener will be lost.

[sjyangkevin]

I will get back to this early next week due to limited bandwidth.