vuonghoangbntt opened a new pull request, #73433:
URL: https://github.com/apache/airflow/pull/73433
closes: #73432
Once a Dag-level callback has run for a file, the Dag processor reparses
that file on
every loop and stops honouring `[dag_processor] min_file_process_interval`
for it. It
doesn't clear on its own. Restarting the processor fixes it, until the next
Dag run
failure.
### Why
`_add_callback_to_queue` is the only place that sets `bundle_version` on a
`DagFileInfo`; the bundle scan leaves it unset. That's deliberate.
`bundle_path` is
`compare=False`, so `bundle_version` is the only field keeping a
version-pinned callback
distinct from the scan entry. Without it, a pinned callback would parse
against the
tracking checkout instead of the version its Dag run used.
The side effect is that one file can hold two `_file_stats` entries.
`prepare_file_queue` collapses them with
```python
file_stats_by_presence_key = {file.presence_key: stat for file, stat in
self._file_stats.items()}
```
which keeps whichever went in last. That's the callback entry, since the
file must be
parsed before a run can fail.
That entry never gets a timestamp. `process_parse_results` builds a
callback-only stat
without `last_finish_time` (deliberately, to avoid disturbing 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.
It stays invisible in the stats table, because `_log_file_processing_stats`
reads
`_file_stats` by exact key. It reports the scan entry's healthy timestamp
while the
queue logic reads the frozen one.
Only versioning-capable bundles hit this. For unversioned requests
`request.bundle_version` is `None` (see #72930), the two keys coincide, and
the merge is
correct.
### Fix
Resolve the `presence_key` collapse by most recent parse rather than by
insertion order.
`_sort_by_mtime` and `processed_recently` now share the helper; they used to
disagree
with each other on which duplicate wins, last match against first match.
This leaves identity and lifecycle alone, so the behaviour #66484 relies on
still holds:
a versioned callback entry survives cleanup while the unversioned file is
present. Using
`>=` keeps today's last-wins result when both candidates are unparsed, so
nothing
changes outside the bug.
### Reproduction
Local Airflow built from main, sqlite, `GitDagBundle` (versioning on),
`min_file_process_interval = 30`, two Dags: one with a Dag-level
`on_failure_callback`
and a task that raises, one control. Measured by polling
`dag.last_parsed_time`:
| 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 |
| after this change, fresh callback fired | 1 per 30.0s | 1 per 30.0s |
The third row triggers a fresh failure after the restart. A restart on its
own clears
the state, so measuring that alone would prove nothing.
I found this on 3.3.1, where it held a Dag processor at its 1 CPU limit
indefinitely.
Two Dag files were reparsing about 43x more often than configured while the
other five
in the same bundle sat exactly on interval.
### Tests
Three regression tests, one per changed lookup site. Each seeds both a scan
entry and a
stale callback entry for the same `presence_key`, which no existing test
does; the
current versioned-stat tests all seed only the versioned entry. I checked
they fail
against unpatched `manager.py` and pass with the change.
### Verification run locally
- New tests: 3 passed with the change, 3 failed without it.
- `airflow-core/tests/unit/dag_processing/test_manager.py`: 173 passed, 5
skipped
(postgres-only). 5 errors at fixture setup come from a gap in my local env
(`ModuleNotFoundError: No module named 'airflow_shared'` in
`cap_structlog`). I get the
same 5 with and without the change, so they are unrelated.
- `ruff check` and `ruff format --check` are clean, and so is `mypy` on the
changed
module.
--
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]
