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
)
            self.executor.queue_command(
                ti,
                command,
                priority=priority,
                queue=queue,
            )
    def _critical_section_execute_task_instances(self, session: Session) -> int:
        """
        Attempts to execute TaskInstances that should be executed by the scheduler.
        There are three steps:
        1. Pick TIs by priority with the constraint that they are in the expected states
        and that we do exceed max_active_runs or pool limits.
        2. Change the state for the TIs above atomically.
        3. Enqueue the TIs in the executor.
        HA note: This function is a "critical section" meaning that only a single executor process can execute
        this function at the same time. This is achieved by doing ``SELECT ... from pool FOR UPDATE``. For DBs
        that support NOWAIT, a "blocked" scheduler will skip this and continue on with other tasks (creating
        new DAG runs, progressing TIs from None to SCHEDULED etc.); DBs that don't support this (such as
        MariaDB or MySQL 5.x) the other schedulers will wait for the lock before continuing.
        :param session:
        :type session: sqlalchemy.orm.Session
        :return: Number of task instance with state changed.
        """
        max_tis = min(self.max_tis_per_query, self.executor.slots_available)
        queued_tis = self._executable_task_instances_to_queued(max_tis, session=session)
        self._enqueue_task_instances_with_queued_state(queued_tis)
        return len(queued_tis)
    @provide_session
    def _change_state_for_tasks_failed_to_execute(self, session: Session = None):