dag processor manager, add retry_db_transcation to _fetch_callbacks (#…

…30079)

* dag processor manager, add retry_db_transcation to _fetch_callbacks

* dag processor manager, create separate method to fetch callbacks with retries to split up provide_session and retry_db_transaction

airflow/dag_processing/manager.py

@@ -62,6 +62,7 @@
     reap_process_group,
     set_new_process_group,
 )
+from airflow.utils.retries import retry_db_transaction
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.sqlalchemy import prohibit_commit, skip_locked, with_row_locks
 
@@ -687,6 +688,10 @@ def _run_parsing_loop(self):
 
     @provide_session
     def _fetch_callbacks(self, max_callbacks: int, session: Session = NEW_SESSION):
+        self._fetch_callbacks_with_retries(max_callbacks, session)
+
+    @retry_db_transaction
+    def _fetch_callbacks_with_retries(self, max_callbacks: int, session: Session):
         """Fetches callbacks from database and add them to the internal queue for execution."""
         self.log.debug("Fetching callbacks from the database.")
         with prohibit_commit(session) as guard: