Assets updated by a dag are attached to the last task in a dag graph rather than the task that creates the update

[TJaniF]

Apache Airflow version

3.0.0

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

No response

What happened?

Running this dag:

from airflow.sdk import Asset, dag, task, chain


@dag
def my_dag():

    @task(outlets=[Asset("my_asset")])
    def task_1():
        return "Task 1 completed"

    @task
    def task_2():
        return "Task 2 completed"

    chain(task_1(), task_2())


my_dag()

The asset is attached to task_2 instead of task_1 in the graph:

What you think should happen instead?

In Airflow 2 the equivalent dag has the asset attached to task_1:

How to reproduce

Run the dag, view the graph.

Operating System

MacOs

Versions of Apache Airflow Providers

breeze

Deployment

Other

Deployment details

breeze

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

[bbovenzi]

@pierrejeambrun Another part of the external_dependencies we should fix after 3.0.0. Although the task->asset info isnt saved in DagDependencies right now. So we might need to do an additional lookup

[phanikumv]

@pierrejeambrun did you get a chance to look into this yet?

[pierrejeambrun]

I did not. Moved it up in my todo list, I'll get to early this week.