https://github.com/apache/airflow/blob/main/airflow/dag_processing/manager.py#L773
if isinstance(request, SlaCallbackRequest):
            if request in self._callback_to_execute[request.full_filepath]:
                self.log.debug("Skipping already queued SlaCallbackRequest")
                return
            
            self.log.debug("Queuing SlaCallbackRequest for %s", request.dag_id)
            self._callback_to_execute[request.full_filepath].append(request)
            if request.full_filepath not in self._file_path_queue:
                self._file_path_queue.append(request.full_filepath)
        
        
        else:
            self.log.debug("Queuing %s CallbackRequest: %s", type(request).__name__, request)
            self._callback_to_execute[request.full_filepath].append(request)
            if request.full_filepath in self._file_path_queue:
                
                
                
                self._file_path_queue = collections.deque(
                    file_path for file_path in self._file_path_queue if file_path != request.full_filepath
                )
            self._file_path_queue.appendleft(request.full_filepath)
    def _refresh_dag_dir(self):