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