nickstenning commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2834891715
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1160,13 +1164,34 @@ def _kill_timed_out_processors(self):
processor = self._processors.pop(proc)
processor.logger_filehandle.close()
- def _add_files_to_queue(self, files: list[DagFileInfo], add_at_front:
bool):
+ def _add_files_to_queue(
+ self,
+ files: list[DagFileInfo],
+ *,
+ mode: Literal["front", "back", "frontprio"],
+ ):
"""Add stuff to the back or front of the file queue, unless it's
already present."""
- new_files = list(f for f in files if f not in self._file_queue)
- if add_at_front:
+ if mode == "frontprio":
+ for file in files:
+ # Try removing the file if already present
+ with contextlib.suppress(ValueError):
+ self._file_queue.remove(file)
+ # enqueue file to the start of the queue.
+ self._file_queue.appendleft(file)
+ elif mode == "front":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extendleft(new_files)
- else:
+ elif mode == "back":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extend(new_files)
+ else:
+ assert_never(mode)
+
+ # If we've just added files to the queue for the first time since
metrics were last emitted, reset the
+ # parse time counter.
+ if not self._parsing_start_time and self._file_queue:
Review Comment:
Yes but without inverting the sense of the condition 😉
--
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]
