You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dygger (via GitHub)" <gi...@apache.org> on 2025/06/30 15:09:12 UTC
[I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
dygger opened a new issue, #52574:
URL: https://github.com/apache/airflow/issues/52574
### 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
```python
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
```python
@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
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W closed issue #52574: Outlet Event Extra Data is Empty in Task Instance Success Listener
URL: https://github.com/apache/airflow/issues/52574
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3149060840
> 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.
Have another look into it. I think `get_template_context` refers to the one in `task_runner.py` which is defined under `sdk/execution_time`. I was looking into the `taskinstance.py` which is under `airflow-core/src/airflow/models`.
I added a logging to print the context, and found that the two invocations referring to different OutletEventAccessors. Below is the one when the "producer" dag is triggered, the OutletEventAccessors object is **0x7fe889c50fb0**.
```
[2025-08-03, 23:48:37] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-08-03, 23:48:37] INFO - Filling up the DagBag from /files/dags/outlet_test_dag.py: source="airflow.models.dagbag.DagBag"
[2025-08-03, 23:48:38] INFO - execution_time.task_runner.RuntimeTaskInstance.get_template_context:: chan="stdout": source="task"
[2025-08-03, 23:48:38] INFO - {'dag': <DAG: asset-producer>, 'inlets': [], 'map_index_template': None, 'outlets': [Asset(name='test-asset', uri='test-asset', group='asset', extra={}, watchers=[])], ...., 'outlet_events': <airflow.sdk.execution_time.context.OutletEventAccessors object at 0x7fe889c50fb0>
```
I think here is when the listener is triggered. The OutletEventAccessors is **0x7fe888329820**
```
[2025-08-03, 23:48:38] INFO - on_task_instance_success: produce: source="test_listener"
[2025-08-03, 23:48:38] INFO - on_task_instance_success outlets: [Asset(name='test-asset', uri='test-asset', group='asset', extra={}, watchers=[])]: source="test_listener"
[2025-08-03, 23:48:38] INFO - execution_time.task_runner.RuntimeTaskInstance.get_template_context:: chan="stdout": source="task"
[2025-08-03, 23:48:38] INFO - {'dag': <DAG: asset-producer>, 'inlets': [], 'map_index_template': None, 'outlets': [Asset(name='test-asset', uri='test-asset', group='asset', extra={}, watchers=[])], ... 'outlet_events': <airflow.sdk.execution_time.context.OutletEventAccessors object at 0x7fe888329820>
```
I attempted to add a simple cache as below, but got the following error.
```python
@functools.cache
def get_template_context(self) -> Context:
...
```
<img width="902" height="252" alt="Image" src="https://github.com/user-attachments/assets/303ed000-28ee-41d5-affc-6392fe91df0f" />
If I understand correctly. I think here the hash key for the cache is `self`, which is of type `RuntimeTaskInstance`. So, the same task instance invokes `get_template_context` will return the same context without going through the logic in the functions to construct the context again. If it is an expected behavior. Wondering if I could get some hints on the recommended way to implement the cache.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "dygger (via GitHub)" <gi...@apache.org>.
dygger commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3095991507
@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](https://airflow.apache.org/docs/apache-airflow/stable/howto/listener-plugin.html) section. Both files should be placed in `$AIRFLOW_HOME/plugins/` folder.
### plugins/test_plugin.py
```python
import test_listener
from airflow.plugins_manager import AirflowPlugin
class TestPlugin(AirflowPlugin):
name = "TestPlugin"
listeners = [test_listener]
```
### plugins/test_listener.py
```python
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
```python
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()
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3130449988
After observing the metadata store, I think the issue is because the following:
First, the asset itself doesn't contain any `extra`.
<img width="1155" height="397" alt="Image" src="https://github.com/user-attachments/assets/ed5520a9-a573-4d4f-8022-1f3fcfd70a4d" />
When you do `outlet_events[asset].extra = {"name1": "value1", "nested_obj": {"name2": "value2"}}`, the `extra` is attached to the asset event.
<img width="1155" height="397" alt="Image" src="https://github.com/user-attachments/assets/55ff1005-771b-43a8-abb7-6854879c81b3" />
The `extra` is propagated into the **consumer** DAG, and can be accessed through either `consumed_asset_events` or `triggering_asset_events`.
<img width="864" height="340" alt="Image" src="https://github.com/user-attachments/assets/b4de3aaf-276d-467e-a4b8-63836661c925" />
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.
https://github.com/apache/airflow/blob/0cd0ff0872cea4dffa088a88b54001d157de6e8e/airflow-core/src/airflow/models/taskinstance.py#L1922-L1925
https://github.com/apache/airflow/blob/0cd0ff0872cea4dffa088a88b54001d157de6e8e/airflow-core/src/airflow/utils/context.py#L126-L148
https://github.com/apache/airflow/blob/0cd0ff0872cea4dffa088a88b54001d157de6e8e/task-sdk/src/airflow/sdk/execution_time/context.py#L503-L512
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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3130710324
We will get it back. 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 will be lost.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3138204856
I will get back to this early next week due to limited bandwidth.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "dygger (via GitHub)" <gi...@apache.org>.
dygger commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3077861313
Any chance to review/triage this? Or maybe there is an another way to retrieve the missing `extra` in `on_task_instance_success` listener?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3130538981
I took a quick look at the code. The cause is simple; the task runner does not commit the asset changes to the database 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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3130308827
looking into it
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3130516809
> Asset extra and asset event extra are two different concepts. I don’t think you’re looking at the right things [@sjyangkevin](https://github.com/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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3130642118
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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3019545715
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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3092708453
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?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Outlet Event Extra Data is Empty in Task Instance Success Listener [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on issue #52574:
URL: https://github.com/apache/airflow/issues/52574#issuecomment-3130511163
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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org