DagNotFound - Dag already exists and and it failed one time with dag_id not found

[sktemkar]

Apache Airflow version

Other Airflow 2 version (please specify below)

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

2.6.1

What happened?

DagNotFound - Dag already exists and and it failed one time with dag_id not found. When we clear the task or resubmit it, the execution is success. Is this something related to dag parsing or dag import timeout.
The number of dags in the dags folder is approximately 1100.
The dag has successful runs for long time, but it suddenly failed with below error.

ERROR - Task failed with exception
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/trigger_dagrun.py", line 155, in execute
replace_microseconds=False,
File "/home/airflow/.local/lib/python3.7/site-packages/airflow/api/common/trigger_dag.py", line 128, in trigger_dag
replace_microseconds=replace_microseconds,
File "/home/airflow/.local/lib/python3.7/site-packages/airflow/api/common/trigger_dag.py", line 52, in _trigger_dag
raise DagNotFound(f"Dag id {dag_id} not found")
airflow.exceptions.DagNotFound: Dag id dag_test1 not found

What you think should happen instead?

It should execute as BAU

How to reproduce

This issue is intermittent and it occurred all of a sudden.
The number of dags in the dags folder is approximately 1100.

Operating System

Azure Aks 1.29.7 ubuntu

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

[potiuk]

This is not airflow issue - this is at most a discussion where you need help with your problem. Most likely the solution you chose for DAG synchronisation is not perfect. There seem to be some race condition that your worker does not have the dag file present in the folder it expects. For example if you have custom DAG syncing solution, it might be that you delete files before you recreate them. Or maybe the syncing in a new pod is not complete when your task starts. This - in k8s environment is usually handled by init container that syncs the files before starting the task.

Generally - look at your DAG syncing deployment choice you made and fix it so that you can be 100% sure that DAG files are present when tasks are running.