DAG disappearing from Airflow in case of standalone DAG processor

[kosteev]

Apache Airflow version

Other Airflow 2 version (please specify below)

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

2.7.3

What happened?

If DAG processor is running in a standalone mode with defined --num-runs flag

airflow dag-processor --num-runs 100

Some DAGs disappear from Airflow until DAG processor will run again.

What you think should happen instead?

DAGs shouldn't disappear never.

How to reproduce

• run Airflow 2.7.3 with standalone DAG processor (separate from scheduler) with --num-runs flag
  Example:

airflow dag-processor --num-runs 100

• deploy some DAGs
• after some time, ~20 mins (once DAG processor will parse all DAG files ~50), deploy another DAG (new DAG file)
• after some time observe that DAGs will start to disappear

The issue is, basically, that:

• DAG processor runs until it parses each file exactly num_runs times
  

  airflow/airflow/dag_processing/manager.py

  Line 1245 in 2872d37

  for stat in self._file_stats.values():

• if a new DAG file will be added to the DAG bag during DAG processor running, this new file will have different number of parsing runs than others already existing there
• at some point other DAGs (those which were already in the DAG bag) will stopped to be parsed due to hitting run limit (they will be excluded from parsing loop), see "Last Run" column in the screenshot below
  

  airflow/airflow/dag_processing/manager.py

  Line 1164 in 2872d37

  files_paths_at_run_limit = [

• DAG processor will continue to parse DAGs until each file will be parsed exactly num_runs times
• Due to the fact that DAGs are not parsed for some time, they will be removed from Airflow DAG bag

Operating System

Ubuntu

Versions of Apache Airflow Providers

No response

Deployment

Other

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

[Taragolis]

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

Is it reproduced in latest Airflow (2.9.0 at that moment)?

[potiuk]

Is it reproduced in latest Airflow (2.9.0 at that moment)?

I do not think there was a change in this area since. But it looks plausible and likely not very difficult to fix - so I will mark it as good-first-issue to investigate and fix.

[github-actions]

This issue has been automatically marked as stale because it has been open for 14 days with no response from the author. It will be closed in next 7 days if no further activity occurs from the issue author.

[kosteev]

I didn't test it in Airflow 2.9, but the implementation of the DAG processor at this place of code should be the same, so the issue has to be there as well.

[potiuk]

@kosteev -> maybe a chance to contribute a fix yourself ?

[andyjianzhou]

Hey, is it okay if I get assigned to this issue? Thanks!