DAG import errors from python modules are not cleared

[ktaborski]

Apache Airflow version

3.1.1

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

No response

What happened?

I have airflow deployed in k8s using official helm chart.
This is example directory structure:

dag.py
module/utils.py

Scenario is:
there is parsing error coming from python module (from ex. utils.py), which is imported/used in DAG code (dag.py).
Then import error from module is fixed, but errors are still visible in UI. Restarting dagProcessor helps.

I did short debugging and most likely issue is related to:

airflow/airflow-core/src/airflow/dag_processing/collection.py

Line 429 in 0448569

good_dag_filelocs = {

It only updates, when import error filename is same as dag.fileloc, which is not true, when import error comes from module.

I think, that same problem was reported here:
#39762 (comment)

What you think should happen instead?

DAG import errors from modules should be cleared.

How to reproduce

Steps provided in description

Operating System

EKS bottlerocket image

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

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

[ephraimbuddy]

I have tried the reproduction in both 3.1.1 and main and can't reproduce it. Is there a special thing I need to do to reproduce it?

[ktaborski]

Indeed, I just realized, that module file (utils.py in this ex.) must be detected as DAG file (might_contain_dag) - it should include dag keyword inside.

[ephraimbuddy]

Can you give an example please. Would be good to sort this out

[ktaborski]

Sure,

$ tree
.
├── dag.py
└── module
    ├── __init__.py
    └── utils.py

dag.py:

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

from module.utils import helloWorld

with DAG(dag_id="hello_world"):
    task1 = PythonOperator(
        task_id="hello_world",
        python_callable=helloWorld)

utils.py:

from airflow.sdk import DAG
from airflow.models import Variable

raise ValueError("xxx")

def helloWorld():
    print("hello world")

__init__.py is empty.
This will cause 2 Dag Import Errors (from dag.py and from utils.py)
then remove from utils.py line raise ValueError("xxx")

After refresh, import error from dag.py is removed, but error from utils.py is still visible.