You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "vuonghoangbntt (via GitHub)" <gi...@apache.org> on 2026/09/21 08:16:31 UTC
[PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "hoangvx-cf (via GitHub)" <gi...@apache.org>.
hoangvx-cf commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4376323214
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a stale
+ callback entry mask a more recent scan parse, so compare parse times instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
Done: the fallback is gone, and both tests now assert the file is queued. I renamed them to `test_prepare_file_queue_queues_file_whose_versioned_stat_is_at_run_limit` and `test_prepare_file_queue_queues_file_whose_versioned_stat_is_recent`.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a stale
+ callback entry mask a more recent scan parse, so compare parse times instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
Done: the fallback is gone, and both tests now assert the file is queued. I renamed them to `test_prepare_file_queue_queues_file_whose_versioned_stat_is_at_run_limit` and `test_prepare_file_queue_queues_file_whose_versioned_stat_is_recent`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "khmelevskiy (via GitHub)" <gi...@apache.org>.
khmelevskiy commented on PR #73433:
URL: https://github.com/apache/airflow/pull/73433#issuecomment-6061475173
We're hitting this on 3.3.2 in production: standalone dag-processors (one per team), `GitDagBundle`, `min_file_process_interval = 120`.
Within two and a half days of a redeploy it started on four files across three dag-processors. Each looping file is parsed 2-3 times a second instead of every 120 s, costs 1-1.5 CPU cores, and its child-process log grows by 80-280 MB a day. A restart clears it; the file's next callback brings it back, usually within a day.
One addition to the issue: here the trigger was task-level callbacks, not Dag-level ones. These are the `TaskCallbackRequest`s the scheduler sends for tasks it fails itself. The loop started in the same minute as the file's first callback. The request carries `bundle_version` either way, so this PR covers both.
main still resolves the stats by insertion order in `prepare_file_queue` and `_sort_by_mtime`.
@jedcunningham @ephraimbuddy could you take a look? A backport to v3-3-test would help too.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4376331032
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a stale
+ callback entry mask a more recent scan parse, so compare parse times instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
Done: the fallback is gone, and both tests now assert the file is queued. I renamed them to `test_prepare_file_queue_queues_file_whose_versioned_stat_is_at_run_limit` and `test_prepare_file_queue_queues_file_whose_versioned_stat_is_recent`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4266473084
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1087,6 +1087,61 @@ def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_
assert known_file not in manager._file_stats
assert versioned_file in manager._file_stats
+ @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+ def test_prepare_file_queue_ignores_stale_callback_stats_for_recently_parsed_file(self):
+ """A stale callback stat must not mask a recent parse of the same file."""
+ freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
+ known_file = _get_file_infos(["file_1.py"])[0]
+ versioned_file = _get_versioned_file_info("file_1.py")
+ known_files = {"testing": {known_file}}
+
+ manager = DagFileProcessorManager(max_runs=3)
+ # A callback pins bundle_version, so its entry is added after the scan entry and
+ # then never refreshed -- resolving by insertion order would pick the stale one.
+ manager._file_stats = {
+ known_file: DagFileStat(1, 0, freezed_base_time - timedelta(seconds=5), 1.0, 1, 1),
+ versioned_file: DagFileStat(1, 0, freezed_base_time - timedelta(hours=2), 1.0, 1, 1),
Review Comment:
Thank kaxil, you're right. test_callback_only_parse_does_not_requeue_recently_parsed_file now runs a real callback-only handle_parsing_result, so the None comes from process_parse_results rather than from the test, in both insertion orders.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4247470290
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1560,14 +1578,7 @@ def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
return file_infos, changed_recently
def processed_recently(self, now, file):
Review Comment:
`processed_recently` has had no caller outside tests since #66484 removed it from `prepare_file_queue` (3.3.0). Could we delete it and its two tests rather than update it? `test_processed_recently_uses_newest_stat` doesn't cover the bug, since nothing on the queue path calls this method.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a stale
+ callback entry mask a more recent scan parse, so compare parse times instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
A callback entry only gets a `last_finish_time` when its processor times out in `_kill_timed_out_processors`. So if a callback for an older bundle version times out, that timestamp wins here and the current file waits a full `min_file_process_interval` before its next parse, though nothing parsed the current version. In `modified_time` mode an edit made in that window also waits for the interval to run out. Would it be simpler to prefer the `bundle_version is None` entry and only fall back to a versioned one when there is no scan entry? Not blocking.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1087,6 +1087,61 @@ def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_
assert known_file not in manager._file_stats
assert versioned_file in manager._file_stats
+ @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+ def test_prepare_file_queue_ignores_stale_callback_stats_for_recently_parsed_file(self):
+ """A stale callback stat must not mask a recent parse of the same file."""
+ freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
+ known_file = _get_file_infos(["file_1.py"])[0]
+ versioned_file = _get_versioned_file_info("file_1.py")
+ known_files = {"testing": {known_file}}
+
+ manager = DagFileProcessorManager(max_runs=3)
+ # A callback pins bundle_version, so its entry is added after the scan entry and
+ # then never refreshed -- resolving by insertion order would pick the stale one.
+ manager._file_stats = {
+ known_file: DagFileStat(1, 0, freezed_base_time - timedelta(seconds=5), 1.0, 1, 1),
+ versioned_file: DagFileStat(1, 0, freezed_base_time - timedelta(hours=2), 1.0, 1, 1),
Review Comment:
Production never gives this entry a timestamp: `process_parse_results` builds the callback-only stat with `last_finish_time=None`. All three new tests seed it with `base - 2h` instead, so the `or utc_epoch()` fallback in `_stats_by_presence_key`, which is the branch that fixes the real bug, never runs. Could you seed this entry with `last_finish_time=None`, or parametrize over `None` and an old time? A test that reaches this state through `_add_callback_to_queue` plus a callback-only `handle_parsing_result` would pin the actual sequence.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4376331032
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a stale
+ callback entry mask a more recent scan parse, so compare parse times instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
Done: the fallback is gone, and both tests now assert the file is queued.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4378680143
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1560,14 +1578,7 @@ def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
return file_infos, changed_recently
def processed_recently(self, now, file):
Review Comment:
Open #75670
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4266530593
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a stale
+ callback entry mask a more recent scan parse, so compare parse times instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
Thanks for your sharp eye. _kill_timed_out_processors actually stamps the versioned key with utcnow(), so a timed-out callback could delay a due parse or, in modified_time mode, hide an edit. I've switched to preferring the scan entry; test_timed_out_callback_does_not_delay_due_parse and test_timed_out_callback_does_not_hide_an_edit cover both through the real timeout path and fail on the previous revision.
I've implemented the fallback to a versioned entry as well, but does it add any value here? A callback run never produces a parsing result, so a versioned stat is only ever None or a timeout stamp, and a file in the current bundle always has a scan entry by the time prepare_file_queue looks at it. I've kept it for now because two tests from #66484 rely on it:
- test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit
- test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4376531235
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1560,14 +1578,7 @@ def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
return file_infos, changed_recently
def processed_recently(self, now, file):
Review Comment:
Thanks, I'll open it as a separate PR
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4377700081
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1573,8 +1573,21 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _resolve_stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve one stat per physical file from the bundle scan's entries.
+
+ A callback pins ``bundle_version``, so its entry is separate from the scan's and does
+ not track parses of the current file: a callback-only result has no ``last_finish_time``,
+ and a timed-out callback processor is stamped although nothing parsed the current version.
+ A file with no scan entry has no stat, so it is queued.
+ """
+ return {
+ file.presence_key: stat for file, stat in self._file_stats.items() if file.bundle_version is None
Review Comment:
`max_runs_reached` still iterates every `_file_stats` value, including versioned ones. A callback-only run keeps the versioned entry at `run_count=0`, so `dag-processor --num-runs N` on a versioned bundle never exits once a callback ran. Pre-existing since #66484, but the same rule applies there: `self._resolve_stats_by_presence_key().values()`. Fine as a follow-up if you want to keep this minimal for the backport.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1171,11 +1173,11 @@ def test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats(s
manager.prepare_file_queue(known_files={"testing": {known_file}})
- assert manager._file_queue == OrderedDict()
+ assert manager._file_queue == OrderedDict.fromkeys([known_file])
@conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"})
@mock.patch("airflow.utils.file.os.path.getmtime")
- def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_entries(
+ def test_recently_modified_file_ignores_versioned_stats_without_creating_duplicate_entries(
Review Comment:
Nit: this passes whether the versioned stat is used or ignored, because its `last_finish_time` is older than the mtime. `test_timed_out_callback_does_not_hide_an_edit` covers the real case, so I'd keep a neutral name here.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #73433:
URL: https://github.com/apache/airflow/pull/73433#issuecomment-6501048917
Nice one @vatsrahul1001!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #73433:
URL: https://github.com/apache/airflow/pull/73433#issuecomment-6347708820
- The PR description still describes the newest-parse approach (`>=`, `processed_recently` sharing the helper, one test per lookup site). Could you update it to match the diff?
- The timed-out tests ([L1246](https://github.com/apache/airflow/blob/27e82d53107e7312f29f5516c71293b3eeaa05a8/airflow-core/tests/unit/dag_processing/test_manager.py#L1246), [L1269](https://github.com/apache/airflow/blob/27e82d53107e7312f29f5516c71293b3eeaa05a8/airflow-core/tests/unit/dag_processing/test_manager.py#L1269)) and the callback-only test ([L1223](https://github.com/apache/airflow/blob/27e82d53107e7312f29f5516c71293b3eeaa05a8/airflow-core/tests/unit/dag_processing/test_manager.py#L1223)) pass whether or not the setup wrote the versioned entry, because the helper ignores it once a scan entry exists. Asserting it before `prepare_file_queue` (`mock_kill` called with `SIGKILL` and `last_finish_time` set, or `None` in the callback-only case) keeps them meaningful if the setup drifts.
- @fahadabdul's gauge report checks out: `prepare_file_queue` passes every `_file_stats` value to `emit_metrics` ([L1648](https://github.com/apache/airflow/blob/27e82d53107e7312f29f5516c71293b3eeaa05a8/airflow-core/src/airflow/dag_processing/manager.py#L1648)), so a timed-out versioned callback keeps `dag_processing.import_errors` at 1 while the UI shows 0. Passing `self._resolve_stats_by_presence_key().values()` there fixes it with the helper this PR adds, so it seems worth folding in.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4351788860
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a stale
+ callback entry mask a more recent scan parse, so compare parse times instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
I suggested the fallback, but you're right that it adds nothing. A versioned stat is only ever empty or a timeout stamp, so when the fallback fires (a file with no scan entry yet) it either does nothing or lets another version's timed-out callback hold back the file's first parse, which is the case this PR fixes. I'd drop it and flip `test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit` and `test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats` to assert the file is queued.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1560,14 +1578,7 @@ def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
return file_infos, changed_recently
def processed_recently(self, now, file):
Review Comment:
Separate PR, please. That keeps this one minimal for the backport.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on PR #73433:
URL: https://github.com/apache/airflow/pull/73433#issuecomment-6414499825
Thanks @kaxil and @fahadabdul, all three are in the latest commit:
- The description now matches the diff.
- The tests assert their setup before `prepare_file_queue`
- `emit_metrics` now gets `self._resolve_stats_by_presence_key().values()`, with a test that a timed-out versioned callback no longer adds to `dag_processing.import_errors`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 merged PR #73433:
URL: https://github.com/apache/airflow/pull/73433
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4378578608
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1171,11 +1173,11 @@ def test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats(s
manager.prepare_file_queue(known_files={"testing": {known_file}})
- assert manager._file_queue == OrderedDict()
+ assert manager._file_queue == OrderedDict.fromkeys([known_file])
@conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"})
@mock.patch("airflow.utils.file.os.path.getmtime")
- def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_entries(
+ def test_recently_modified_file_ignores_versioned_stats_without_creating_duplicate_entries(
Review Comment:
Renamed to `test_recently_modified_file_with_versioned_stat_does_not_create_duplicate_entries`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4266547290
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1560,14 +1578,7 @@ def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
return file_infos, changed_recently
def processed_recently(self, now, file):
Review Comment:
Agreed that it has no caller since #66484, and that the test I added didn't cover the bug, so I've dropped that test and reverted my change to the method. This PR no longer touches it.
Should I remove it in this PR, or would it be better to open a separate one?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "fahadabdul (via GitHub)" <gi...@apache.org>.
fahadabdul commented on PR #73433:
URL: https://github.com/apache/airflow/pull/73433#issuecomment-6339492432
A related symptom from the same versioned callback entry, which this PR doesn't seem to cover: `dag_processing.import_errors` stays at 1 when there are no import errors.
**What seems to happen**
1. A callback runs under a `DagFileInfo` with `bundle_version` set, so it gets its own `_file_stats` entry.
2. If that callback run exceeds `dag_file_processor_timeout`, `_kill_timed_out_processors` writes `DagFileStat(import_errors=1, ...)` to the versioned entry. No `ParseImportError` row is written, so the UI and API show 0.
3. `remove_orphaned_file_stats` compares `presence_key`, so the versioned entry survives while the file exists (as intended since #66484).
4. `emit_metrics` sums `import_errors` across all `_file_stats` values, so the gauge keeps reporting 1 until the DAG processor restarts.
In the production case, the stats table showed 0 errors for the same file at the same time, since `_log_file_processing_stats` reads the scan entry by exact key.
**Repro** (Astronomer Runtime build of 3.3.2, `3.3.2+astro.2`, versioned bundle, `dag_file_processor_timeout = 30`):
```python
import time
from datetime import datetime
from airflow.sdk import dag, task
def slow_callback(context):
time.sleep(120)
@dag(start_date=datetime(2026, 1, 1), schedule=None, catchup=False, on_failure_callback=slow_callback)
def ghost_import_error_repro():
@task
def fail():
raise ValueError("fail on purpose")
fail()
ghost_import_error_repro()
```
After one triggered run, the callback was killed after about 30s, and `dag_processing.import_errors` went from 0 to 1. The `import_error` table was empty while the gauge was at 1. The gauge stayed at 1 after a new bundle version was picked up (about 27 minutes on the same DAG processor), and went back to 0 once the DAG processor was replaced.
We first saw this in production on a 3.2.2 build with #66484 backported. The relevant code (`presence_key` cleanup, the timeout path, and `emit_metrics`) is the same in OSS 3.3.2 and on main.
**Possible directions** (not sure which fits best):
- Don't record `import_errors=1` when the timed-out processor was a versioned callback run, since a slow callback isn't an import error.
- Or have `emit_metrics` resolve per `presence_key` the same way this PR now does for the queue checks (preferring the scan entry).
Happy to open a separate issue if you'd prefer to keep this PR focused.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop a stale callback stat from forcing constant Dag file reparsing [airflow]
Posted by "vuonghoangbntt (via GitHub)" <gi...@apache.org>.
vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4378573994
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1573,8 +1573,21 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
+ def _resolve_stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve one stat per physical file from the bundle scan's entries.
+
+ A callback pins ``bundle_version``, so its entry is separate from the scan's and does
+ not track parses of the current file: a callback-only result has no ``last_finish_time``,
+ and a timed-out callback processor is stamped although nothing parsed the current version.
+ A file with no scan entry has no stat, so it is queued.
+ """
+ return {
+ file.presence_key: stat for file, stat in self._file_stats.items() if file.bundle_version is None
Review Comment:
Good catch, thanks. It needs `_resolve_stats_by_presence_key`, so I'll fix it in a follow-up once this one is merged: #75666.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org