This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 72628b5c4f4 Stop a stale callback stat from forcing constant Dag file
reparsing (#73433)
72628b5c4f4 is described below
commit 72628b5c4f4851c7fb4d693b881065f43d164a00
Author: Vuong Hoang <[email protected]>
AuthorDate: Mon Oct 26 23:34:08 2026 +0700
Stop a stale callback stat from forcing constant Dag file reparsing (#73433)
* Stop a stale callback stat from forcing constant Dag file reparsing
A callback pins bundle_version on its DagFileInfo while the bundle scan
leaves it unset, so one file can hold two _file_stats entries. Collapsing
them by presence_key with a dict comprehension keeps whichever was inserted
last, which is the callback entry, and that entry is never refreshed
afterwards because later parses write to the scan key. Its frozen
last_finish_time makes the file look long unparsed, so
min_file_process_interval never excludes it and the processor reparses it on
every loop until it restarts.
Resolve the collapse by the most recent parse instead. _sort_by_mtime and
processed_recently share the same lookup, which previously disagreed with
each other on which duplicate wins.
* Prefer the bundle scan's stat when resolving a file's parse time
Callback runs never record a parse of the current file: a callback-only
stat has no last_finish_time, and a timed-out callback processor is
stamped anyway. Fall back to a versioned stat only when the file has no
scan entry.
* Use only scan stats for queueing and parse metrics
* Use a neutral name for the duplicate-entries test
---
airflow-core/src/airflow/dag_processing/manager.py | 19 +++-
.../tests/unit/dag_processing/test_manager.py | 112 ++++++++++++++++++++-
2 files changed, 123 insertions(+), 8 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index a4a834b9ace..a150b4c1fba 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -1573,8 +1573,21 @@ class DagFileProcessorManager(LoggingMixin):
# 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
+ }
+
def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
- file_stats_by_presence_key = {file.presence_key: stat for file, stat
in self._file_stats.items()}
+ file_stats_by_presence_key = self._resolve_stats_by_presence_key()
files_with_mtime: dict[DagFileInfo, float] = {}
changed_recently = set()
for file in files:
@@ -1624,19 +1637,19 @@ class DagFileProcessorManager(LoggingMixin):
Note this method is only called when the file path queue is empty
"""
+ file_stats_by_presence_key = self._resolve_stats_by_presence_key()
# We only emit metrics after processing all files in the queue. If
`self._parsing_start_time` is None
# when this method is called, no files have yet been added to the
queue so we shouldn't emit metrics.
if self._parsing_start_time is not None:
emit_metrics(
parse_time=time.perf_counter() - self._parsing_start_time,
- dag_file_stats=list(self._file_stats.values()),
+ dag_file_stats=list(file_stats_by_presence_key.values()),
)
self._parsing_start_time = None
# If the file path is already being processed, or if a file was
# processed recently, wait until the next batch
in_progress_keys = {file.presence_key for file in self._processors}
- file_stats_by_presence_key = {file.presence_key: stat for file, stat
in self._file_stats.items()}
now = timezone.utcnow()
# Sort the file paths by the parsing order mode
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 322d7dd83da..275422c78e4 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -1147,7 +1147,8 @@ class TestDagFileProcessorManager:
assert manager._file_queue == OrderedDict()
@conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
- def
test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit(self):
+ def test_prepare_file_queue_ignores_versioned_stat_at_run_limit(self):
+ """A callback's versioned stat doesn't stand for a parse, so it
doesn't count towards the run limit."""
manager = DagFileProcessorManager(max_runs=1)
versioned_file = _get_versioned_file_info("file_1.py")
known_file = _get_file_infos(["file_1.py"])[0]
@@ -1156,10 +1157,11 @@ class TestDagFileProcessorManager:
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"): "alphabetical"})
- def
test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats(self):
+ def test_prepare_file_queue_ignores_recent_versioned_stat(self):
+ """A callback's versioned stat doesn't stand for a parse, so a recent
one doesn't hold the file back."""
manager = DagFileProcessorManager(max_runs=3)
versioned_file = _get_versioned_file_info("file_1.py")
known_file = _get_file_infos(["file_1.py"])[0]
@@ -1171,11 +1173,11 @@ class TestDagFileProcessorManager:
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_with_versioned_stat_does_not_create_duplicate_entries(
self, mock_getmtime
):
freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
@@ -1197,6 +1199,106 @@ class TestDagFileProcessorManager:
assert known_file not in manager._file_stats
assert versioned_file in manager._file_stats
+ @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+ @pytest.mark.parametrize(
+ "callback_first",
+ [
+ pytest.param(False, id="scan-recorded-first"),
+ pytest.param(True, id="callback-recorded-first"),
+ ],
+ )
+ def test_callback_only_parse_does_not_requeue_recently_parsed_file(self,
callback_first):
+ """A callback-only parse records no ``last_finish_time``, which must
not requeue the 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")
+ scan_stat = DagFileStat(1, 0, freezed_base_time -
timedelta(seconds=5), 1.0, 1, 1)
+
+ manager = DagFileProcessorManager(max_runs=3)
+ processor, _ = self.mock_processor(start_time=time.monotonic() - 1)
+ processor.had_callbacks = True
+ processor.parsing_result = None
+
+ with time_machine.travel(freezed_base_time):
+ if not callback_first:
+ manager._file_stats[known_file] = scan_stat
+ manager.handle_parsing_result(versioned_file, processor)
+ assert manager._file_stats[versioned_file].last_finish_time is None
+ if callback_first:
+ manager._file_stats[known_file] = scan_stat
+ manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+ assert manager._file_queue == OrderedDict()
+
+ @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+ @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True)
+ def test_timed_out_callback_does_not_delay_due_parse(self, mock_kill):
+ """A callback processor that times out must not make a due file look
recently parsed."""
+ 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")
+
+ manager = DagFileProcessorManager(max_runs=3, processor_timeout=5)
+ manager._file_stats[known_file] = DagFileStat(
+ 1, 0, freezed_base_time - timedelta(seconds=60), 1.0, 1, 1
+ )
+ processor, _ = self.mock_processor(start_time=time.monotonic() -
manager.processor_timeout - 1)
+ manager._processors = {versioned_file: processor}
+
+ with time_machine.travel(freezed_base_time):
+ manager._kill_timed_out_processors()
+ mock_kill.assert_called_once_with(processor, signal.SIGKILL)
+ assert manager._file_stats[versioned_file].last_finish_time is not
None
+ manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+ assert known_file in manager._file_queue
+
+ @conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"})
+ @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True)
+ @mock.patch("airflow.utils.file.os.path.getmtime")
+ def test_timed_out_callback_does_not_hide_an_edit(self, mock_getmtime,
mock_kill):
+ """An edit after the last parse is picked up even if a callback
processor times out later."""
+ 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")
+
+ manager = DagFileProcessorManager(max_runs=3, processor_timeout=5)
+ manager._file_stats[known_file] = DagFileStat(
+ 1, 0, freezed_base_time - timedelta(seconds=10), 1.0, 1, 1
+ )
+ processor, _ = self.mock_processor(start_time=time.monotonic() -
manager.processor_timeout - 1)
+ manager._processors = {versioned_file: processor}
+ mock_getmtime.return_value = (freezed_base_time -
timedelta(seconds=7)).timestamp()
+
+ with time_machine.travel(freezed_base_time - timedelta(seconds=5)):
+ manager._kill_timed_out_processors()
+ mock_kill.assert_called_once_with(processor, signal.SIGKILL)
+ assert manager._file_stats[versioned_file].last_finish_time is not None
+ with time_machine.travel(freezed_base_time):
+ manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+ assert known_file in manager._file_queue
+
+ @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+ @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True)
+ @mock.patch("airflow.dag_processing.manager.stats.gauge")
+ def test_timed_out_callback_is_not_reported_as_an_import_error(self,
mock_gauge, mock_kill):
+ """The ``import_errors=1`` a timeout leaves on a callback's entry must
not reach the gauge."""
+ known_file = _get_file_infos(["file_1.py"])[0]
+ versioned_file = _get_versioned_file_info("file_1.py")
+
+ manager = DagFileProcessorManager(max_runs=3, processor_timeout=5)
+ manager._file_stats[known_file] = DagFileStat(1, 2,
timezone.datetime(2020, 1, 5), 1.0, 1, 1)
+ processor, _ = self.mock_processor(start_time=time.monotonic() -
manager.processor_timeout - 1)
+ manager._processors = {versioned_file: processor}
+
+ manager._kill_timed_out_processors()
+ assert manager._file_stats[versioned_file].import_errors == 1
+ manager._parsing_start_time = time.perf_counter()
+ manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+ mock_gauge.assert_any_call("dag_processing.import_errors", 2)
+
def test_file_paths_in_queue_sorted_by_priority(self):
from airflow.models.dagbag import DagPriorityParsingRequest
