https://github.com/apache/airflow/blob/main/airflow/dag_processing/manager.py#L773
files_paths_to_queue = [
            file_path for file_path in file_paths if file_path not in file_paths_to_exclude
        ]
        for file_path, processor in self._processors.items():
            self.log.debug(
                "File path %s is still being processed (started: %s)",
                processor.file_path,
                processor.start_time.isoformat(),
            )
        self.log.debug("Queuing the following files for processing:\n\t%s", "\n\t".join(files_paths_to_queue))
        default = DagFileStat(
            num_dags=0, import_errors=0, last_finish_time=None, last_duration=None, run_count=0
        )
        for file_path in files_paths_to_queue:
            self._file_stats.setdefault(file_path, default)
        self._file_path_queue.extend(files_paths_to_queue)
    def _kill_timed_out_processors(self):
        """Kill any file processors that timeout to defend against process hangs."""
        now = timezone.utcnow()
        processors_to_remove = []
        for file_path, processor in self._processors.items():
            duration = now - processor.start_time
            if duration > self._processor_timeout:
                self.log.error(
                    "Processor for %s with PID %s started at %s has timed out, killing it.",
                    file_path,
                    processor.pid,
                    processor.start_time.isoformat(),
                )