blinkseb opened a new pull request, #30392: URL: https://github.com/apache/airflow/pull/30392
<!-- Thank you for contributing! Please make sure that your code changes are covered with tests. And in case of new features or big changes remember to adjust the documentation. Feel free to ping committers for the review! In case of an existing issue, reference it using one of the following: closes: #ISSUE related: #ISSUE How to write a good git commit message: http://chris.beams.io/posts/git-commit/ --> Hello, While debugging an issue on why our DAGs were running without tasks (issue https://github.com/apache/airflow/issues/18880), I found a possible "race condition" in `DagBag.get_dag`. Since DAGs are serialized and written to database in a different process than the scheduler, it's possible to end up in a situation where the serialized DAG `last_updated` field is before the DagBag `last_fetched` datetime, while the cached version in the DagBag is not the latest one. Once this situation is reached, the only was to refresh the DAG is either to modify it, or to restart the scheduler. The diagram below should help to illustrate the problem: ```mermaid sequenceDiagram DagProcessor->>DagProcessor: Parse and collect DAGs activate DagProcessor Note left of DagProcessor: Begin Transaction DagProcessor->>DagProcessor: Serialize DAG A (last_updated = 10:00:00) DagProcessor->>DagProcessor: serialize DAG B (last_updated = 10:00:01) DagProcessor->>DagProcessor: serialize DAG C (last_updated = 10:00:02) par DagProcessor->>DagProcessor: serialize DAG D (last_updated = 10:00:03) and Scheduler->>Scheduler: read DAG A (last_fetched = 10:00:03) Note right of Scheduler: previous version of DAG A was fetched end DagProcessor->>DagProcessor: serialize DAG E (last_updated = 10:00:04) Note left of DagProcessor: End Transaction deactivate DagProcessor ``` If you have a lot of complex DAGs in a single file, the serialization step can take a long time. If, during this time, the scheduler has to run a DAG currently being serialized, you can have a situation where the `last_fetched` value of the DagBag is after the `last_updated` value of the serialized DAG while still having the previous version of the DAG in memory. This PR fixes the issue by also checking the serialized DAG hash in addition to the last updated datetime. I'm wondering if maybe we can just use the hash here instead of the updated datetime, but I'm not sure of the possible side-effects. I've added a new unit tests replicating the issue. Without the patch, the test fails. Thanks, --- **^ Add meaningful description above** Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information. In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed. In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x). In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
