Repository navigation
Fix #73432: preserve last_finish_time in callback-only DagFileProcessorManager processing - #73771
Closed
chinmayasameeru wants to merge 1 commit into
Closed
chinmayasameeru wants to merge 1 commit into
chinmayasameeru wants to merge 1 commit into
Conversation
…rocessorManager processing When a DagFileProcessorManager processes a file in callback-only mode (is_callback_only=True), process_parse_results creates a new DagFileStat with last_finish_time=None. This new stat shadows the existing stat's timestamp in _file_stats, causing the file to appear as "never parsed" on subsequent scans. The file gets re-queued every scan cycle, creating an infinite loop. The fix adds a last_finish_time parameter to process_parse_results. In the is_callback_only branch, the passed last_finish_time (from the existing stat) is preserved instead of using the current finish_time. The handle_parsing_result call site passes self._file_stats[file].last_finish_time, and the process_unparsed_files path passes db_stat.last_finish_time to DagFileStat.from_db. Closes apache#73432
chinmayasameeru
requested review from
ephraimbuddy and
jedcunningham
as code owners
September 27, 2026 07:27
|
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
|
Eason09053360
left a comment
Contributor
There was a problem hiding this comment.
This duplicates #73433 (opened earlier by the issue reporter).
Also, ruff format fails on the function signature — please run prek install so the static checks run before pushing. :)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fix #73432 — the dag processor can get stuck in an infinite loop after a Dag-level callback runs.
Here's what happens: when a Dag file gets processed in "callback-only" mode (meaning it was triggered by a Dag callback, not a regular scan),
process_parse_resultscreates a newDagFileStatwithlast_finish_time=None. That stat overwrites the existing one in_file_stats, so on the next scan the file looks like it's never been parsed. The scheduler re-queues it, processes it again, creates another stat withlast_finish_time=None, and the cycle repeats forever.The fix is to preserve the existing
last_finish_timewhen doing callback-only processing. I added alast_finish_timeparameter toprocess_parse_results— when it's passed (from the existing stat), theis_callback_onlybranch uses it instead of the currentfinish_time. I updated both call sites:handle_parsing_result: passesself._file_stats[file].last_finish_time— this is the main fix for the reported issueprocess_unparsed_files: passesdb_stat.last_finish_timetoDagFileStat.from_db— this handles the case where a file is already in the DBI couldn't run the full airflow test suite here (the test environment needs a database and a bunch of infra), but I did validate the Python syntax with
py_compileand the logic is pretty straightforward — it's just passing a timestamp through instead of dropping it.