Removes AIP-44 from dag processing manager and internal_api_call (apa…

…che#44549)

This is the last "internal_api_call" change from AIP-44 removal
and it not only removes the dag_processsing/manager calls but also
the internal_api_call itself.

Part of apache#44436

airflow/dag_processing/manager.py

@@ -42,7 +42,6 @@
 from tabulate import tabulate
 
 import airflow.models
-from airflow.api_internal.internal_api_call import internal_api_call
 from airflow.callbacks.callback_requests import CallbackRequest
 from airflow.configuration import conf
 from airflow.dag_processing.processor import DagFileProcessorProcess
@@ -499,7 +498,6 @@ def _scan_stale_dags(self):
             self.last_deactivate_stale_dags_time = timezone.utcnow()
 
     @classmethod
-    @internal_api_call
     @provide_session
     def deactivate_stale_dags(
         cls,
@@ -698,23 +696,14 @@ def _run_parsing_loop(self):
                         poll_time = 0.0
 
     @classmethod
-    @internal_api_call
     @provide_session
+    @retry_db_transaction
     def _fetch_callbacks(
         cls,
         max_callbacks: int,
         standalone_dag_processor: bool,
         dag_directory: str,
         session: Session = NEW_SESSION,
-    ) -> list[CallbackRequest]:
-        return cls._fetch_callbacks_with_retries(
-            max_callbacks, standalone_dag_processor, dag_directory, session
-        )
-
-    @classmethod
-    @retry_db_transaction
-    def _fetch_callbacks_with_retries(
-        cls, max_callbacks: int, standalone_dag_processor: bool, dag_directory: str, session: Session
     ) -> list[CallbackRequest]:
         """Fetch callbacks from database and add them to the internal queue for execution."""
         cls.logger().debug("Fetching callbacks from the database.")
@@ -765,7 +754,6 @@ def _refresh_requested_filelocs(self) -> None:
             self._file_path_queue.appendleft(fileloc)
 
     @classmethod
-    @internal_api_call
     @provide_session
     def _get_priority_filelocs(cls, session: Session = NEW_SESSION):
         """Get filelocs from DB table."""
@@ -831,7 +819,6 @@ def _print_stat(self):
             self.last_stat_print_time = time.monotonic()
 
     @staticmethod
-    @internal_api_call
     @provide_session
     def clear_nonexistent_import_errors(
         file_paths: list[str] | None, processor_subdir: str | None, session=NEW_SESSION