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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
