vuonghoangbntt opened a new issue, #73432:
URL: https://github.com/apache/airflow/issues/73432
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.3.1
### What happened and how to reproduce it?
### Issue Description
A Dag with a Dag-level `on_failure_callback` parses on schedule until one of
its runs
fails. After that the Dag processor reparses that file on every loop and
stops applying
`[dag_processor] min_file_process_interval` to it. It never recovers on its
own.
Restarting the Dag processor clears it. The next failed run brings it
straight back.
We hit this in production on 3.3.1, where it held a Dag processor at its 1
CPU limit for
days. Two Dag files were reparsing roughly 43x more often than configured.
The other
five files in the same bundle stayed exactly on interval.
The `DAG File Processing Stats` table gives no hint. It shows recent, healthy
`Last Run At` values for the affected file the whole time.
### Root cause
Of the five `DagFileInfo(...)` construction sites, only
`_add_callback_to_queue` sets
`bundle_version`. The bundle scan leaves it unset, and that's on purpose:
`bundle_path`
is `field(compare=False)`, so `bundle_version` is the only field that keeps a
version-pinned callback distinct from the scan entry. Drop it and a pinned
callback
parses against the tracking checkout instead of the version its Dag run used.
The side effect is that one physical file can end up holding two
`_file_stats` entries.
`prepare_file_queue` collapses them:
```python
file_stats_by_presence_key = {file.presence_key: stat for file, stat in
self._file_stats.items()}
```
A dict comprehension keeps whichever key went in last. That is the callback
entry, since
the file has to be parsed before a run can fail, so it shadows the scan
entry.
That entry never gets a timestamp. `process_parse_results` builds a
callback-only stat
without `last_finish_time` (deliberately, so it does not disturb stale-Dag
detection),
and `handle_parsing_result` writes it over the entry wholesale, so the value
is `None`.
It stays `None`, because every later parse writes to the scan key instead.
From then on
the interval check sees a file that has apparently never been parsed, and
requeues it
every loop.
Nothing cleans it up, either. `remove_orphaned_file_stats` matches on
`presence_key`,
which is still present.
Two nearby places resolve the same lookup and disagree with each other. The
identical
collapse is duplicated in `_sort_by_mtime`, which feeds `changed_recently`,
the other
route that bypasses the interval. And `processed_recently` resolves it by
first match,
the opposite tiebreak.
What keeps it hidden is `_log_file_processing_stats`, which reads
`_file_stats` by exact
key. It reports the scan entry's healthy timestamp while the queue logic
reads the
frozen one.
### Scope
This only bites bundles that support versioning. When a request carries no
version,
`request.bundle_version` is `None` (since #72930), the two keys coincide,
and merging
them is correct.
### Steps to reproduce
1. Airflow with a standalone `dag-processor` and a versioning bundle
(`GitDagBundle`),
so the bundle reports a non-null version.
2. Set `[dag_processor] min_file_process_interval = 30`.
3. Add two Dags to the bundle. One has a **Dag-level** `on_failure_callback`
and a task
that raises; the other is a control with no callback:
```python
with DAG(dag_id="callback_dag", schedule=None, on_failure_callback=_notify):
PythonOperator(task_id="boom", python_callable=_boom) # raises
```
4. Trigger `callback_dag` and let the run fail, so the Dag-level callback is
dispatched.
5. Watch the reparse rate, e.g. by polling `dag.last_parsed_time`, or the
statsd counter
`dag_processing.last_duration.<file>`.
Measured on main (sqlite, `GitDagBundle`, `min_file_process_interval = 30`):
| phase | callback_dag | control_dag |
|------------------------------------|--------------|-------------|
| baseline, before any callback | 1 per 30.0s | 1 per 30.0s |
| after the Dag-level callback fired | 1 per 0.4s | 1 per 30.0s |
Restarting the Dag processor clears it, because the bad state lives in
memory. The next
failed Dag run re-arms it.
### What you think should happen instead?
min_file_process_interval` should keep applying to a file whether or not a
Dag-level
callback has ever run for it. One callback should not change that file's
parse cadence
permanently.
### Operating System
_No response_
### Deployment
Official Apache Airflow Helm Chart
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
1.22.0 (latest released)
### Kubernetes Version
1.34.9
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
