AirflowInactiveAssetInInletOrOutletException is flaky

[kacpermuda]

Apache Airflow version

3.1.2

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

Happens since 3.0

What happened?

When running DAG (this) that has tasks with inlets and outlets that are not consumed anywhere the AirflowInactiveAssetInInletOrOutletException is raised, but only sometimes, every few days.

These are my tests that run daily, and you can see that these failures look rather random (task2 is the one with "inactive" inlets). Also, whenever a failure happens and I create a new DagRun straight away, it passes without error.

Inlets/outlets are not consumed - it means that inlets to a single task are not outlets of any other task in my entire Airflow, the same for outlets - they are not inlets to any other task in my entire Airflow. I thought it's not a requirement, a task should be able to emit Asset regardless. These inlets and outlets are used in tasks so that I can check how the OpenLineage is serializing them into OL events, so I can't simply remove them.

What you think should happen instead?

I think no error should be raised and I should be able to define inlets/outlets to my tasks that are not defined anywhere else. If, however, It's considered a wrong behavior, errors should always be raised so it's consistent.

How to reproduce

That's a hard one to reproduce quickly. I could say: run a DAG with inlet that has some non-existing uri, every day for a week, and see if it raises an error.

I assume maybe it's "flakiness" comes from some internal process that clears AssetActive table or something similar? Really hard to say, I tried to follow the code, ended up here in API's validate_inlets_and_outlets, but I still can't figure out why my inlet is not in the AssetActive table (maybe it's not being inserted properly when dag is parsed?)

Operating System

linux

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

Running on Airflow 3.0 and 3.1, it has been happening since the release of AF3 really, so I'm not sure if any recent change is the cause of it.

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

[mykola-shyshov]

let me check. it is not hard to reproduce.

from datetime import datetime, timedelta
from airflow import DAG, Asset
from airflow.decorators import task

# Create an asset with a non-existing URI
non_existing_asset = Asset("s3://non-existing-bucket/path/to/data")

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2024, 1, 1),
    "retries": 0,
}

with DAG(
    dag_id="reproduce_issue_58303",
    default_args=default_args,
    description="Reproduce flaky AirflowInactiveAssetInInletOrOutletException",
    schedule=timedelta(seconds=30),  # Run every 30 seconds to trigger race condition
    catchup=False,
    tags=["bug_reproduction", "issue_58303"],
) as dag:

    @task(
        inlets=[non_existing_asset],  # This should trigger the flaky exception
        outlets=[],
    )
    def task_with_unused_inlet():
        """
        Simple task with an inlet that references a non-existing asset.
        According to issue #58303, this should intermittently raise
        AirflowInactiveAssetInInletOrOutletException due to a race condition.
        """
        print("Task executed successfully")
        print(f"This task declares inlet: {non_existing_asset.uri}")
        return "success"

    task_with_unused_inlet()

[mykola-shyshov]

scheduler deactivate assets in case they are not in outlet and are not referenced by DAG

Check assets orphanization and update their active entry. An orphaned asset is no longer referenced in any DAG schedule parameters
or task outlets. Active assets (non-orphaned) have entries in AssetActive
and must have unique names and URIs.

This issue is relevant for DAGs scheduled by time.
I think the next fix will work

        orphaned = (
            (func.count(DagScheduleAssetReference.dag_id) + func.count(TaskOutletAssetReference.dag_id) + func.count(TaskInletAssetReference.dag_id)) == 0
        ).label("orphaned")
        asset_reference_query = session.execute(
            select(orphaned, AssetModel)
            .outerjoin(DagScheduleAssetReference)
            .outerjoin(TaskOutletAssetReference)
            .outerjoin(TaskInletAssetReference)
            .group_by(AssetModel.id)
            .order_by(orphaned)
        )

will prepare PR

[kacpermuda]

Great, thanks for debugging this !