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