We have the first example of scheduler deadlock in the willd for MySQL for the new scheduler:

https://github.com/apache/airflow/pull/8550/checks?check_run_id=1258193958#step:6:1439

Would be great to track it down and fix before the prod release:

**NOTE**  For now the issue has been mitigated (according to the documentation of MySQL) by applying restarts (with exponential back-off) in #12046. This should help with beta-testing of Airlfow by the users and to see if the retries on deadlock has some potential negative effects (according to the documentation - they might have if they happen frequently enough). 

Some more analysis might be performed between now and 2.0.0 which might result in deciding that this issue does not need any more fixes than the current mitigation or some actual changes in the code that might result in preventing the deadlocks.


airflow/jobs/scheduler_job.py
:param parent_channel: the parent end of the channel to close in the child
        :type result_channel: multiprocessing.Connection
        :param file_path: the file to process
        :type file_path: str
        :param pickle_dags: whether to pickle the DAGs found in the file and
            save them to the DB
        :type pickle_dags: bool
        :param dag_ids: if specified, only examine DAG ID's that are
            in this list
        :type dag_ids: list[str]
        :param thread_name: the name to use for the process that is launched
        :type thread_name: str
        :param callback_requests: failure callback to execute
        :type callback_requests: List[airflow.utils.callback_requests.CallbackRequest]
        :return: the process that was launched
        :rtype: multiprocessing.Process
        """
        
        log: logging.Logger = logging.getLogger("airflow.processor")
        
        
        log.info("Closing parent pipe")
        parent_channel.close()
        del parent_channel
        set_context(log, file_path)
        setproctitle("airflow scheduler - DagFileProcessor {}".format(file_path))
        try:
            
            with ExitStack() as exit_stack:
                exit_stack.enter_context(redirect_stdout(StreamLogWriter(log, logging.INFO)))