
For example here:

https://github.com/apache/airflow/runs/2288380184?check_suite_focus=true#step:6:8759

```

 =================================== FAILURES ===================================
  __________ TestSchedulerJob.test_scheduler_keeps_scheduling_pool_full __________
  
  self = <tests.jobs.test_scheduler_job.TestSchedulerJob testMethod=test_scheduler_keeps_scheduling_pool_full>
  
      def test_scheduler_keeps_scheduling_pool_full(self):
          """
          Test task instances in a pool that isn't full keep getting scheduled even when a pool is full.
          """
          dag_d1 = DAG(dag_id='test_scheduler_keeps_scheduling_pool_full_d1', start_date=DEFAULT_DATE)
          BashOperator(
              task_id='test_scheduler_keeps_scheduling_pool_full_t1',
              dag=dag_d1,
              owner='airflow',
              pool='test_scheduler_keeps_scheduling_pool_full_p1',
              bash_command='echo hi',
          )
      
          dag_d2 = DAG(dag_id='test_scheduler_keeps_scheduling_pool_full_d2', start_date=DEFAULT_DATE)
          BashOperator(
              task_id='test_scheduler_keeps_scheduling_pool_full_t2',
              dag=dag_d2,
              owner='airflow',
              pool='test_scheduler_keeps_scheduling_pool_full_p2',
              bash_command='echo hi',
          )
          dagbag = DagBag(
              dag_folder=os.path.join(settings.DAGS_FOLDER, "no_dags.py"),
              include_examples=False,
              read_dags_from_db=True,
          )
          dagbag.bag_dag(dag=dag_d1, root_dag=dag_d1)
          dagbag.bag_dag(dag=dag_d2, root_dag=dag_d2)
          dagbag.sync_to_db()
      
          session = settings.Session()
          pool_p1 = Pool(pool='test_scheduler_keeps_scheduling_pool_full_p1', slots=1)
          pool_p2 = Pool(pool='test_scheduler_keeps_scheduling_pool_full_p2', slots=10)
          session.add(pool_p1)
          session.add(pool_p2)
          session.commit()
      
          dag_d1 = SerializedDAG.from_dict(SerializedDAG.to_dict(dag_d1))
      
          scheduler = SchedulerJob(executor=self.null_exec)
          scheduler.processor_agent = mock.MagicMock()
      
          # Create 5 dagruns for each DAG.
          # To increase the chances the TIs from the "full" pool will get retrieved first, we schedule all
          # TIs from the first dag first.
          date = DEFAULT_DATE
          for _ in range(5):
              dr = dag_d1.create_dagrun(
                  run_type=DagRunType.SCHEDULED,
                  execution_date=date,
                  state=State.RUNNING,
              )
              scheduler._schedule_dag_run(dr, {}, session)
              date = dag_d1.following_schedule(date)
      
          date = DEFAULT_DATE
          for _ in range(5):
              dr = dag_d2.create_dagrun(
                  run_type=DagRunType.SCHEDULED,
                  execution_date=date,
                  state=State.RUNNING,
              )
              scheduler._schedule_dag_run(dr, {}, session)
              date = dag_d2.following_schedule(date)
      
          scheduler._executable_task_instances_to_queued(max_tis=2, session=session)
          task_instances_list2 = scheduler._executable_task_instances_to_queued(max_tis=2, session=session)
      
          # Make sure we get TIs from a non-full pool in the 2nd list
          assert len(task_instances_list2) > 0
  >       assert all(
              task_instance.pool != 'test_scheduler_keeps_scheduling_pool_full_p1'
              for task_instance in task_instances_list2
          )
  E       AssertionError: assert False
  E        +  where False = all(<generator object TestSchedulerJob.test_scheduler_keeps_scheduling_pool_full.<locals>.<genexpr> at 0x7fb6ecb90c10>)
 ``` 
tests/jobs/test_scheduler_job.py
"""
        Test that Operator links are not loaded inside the Scheduling Loop (that does not include
        DagFileProcessorProcess) especially the critical loop of the Scheduler.
        This is to avoid running User code in the Scheduler and prevent any deadlocks
        """
        with dag_maker(dag_id='test_extra_operator_links_not_loaded_in_scheduler') as dag:
            
            _ = CustomOperator(task_id='custom_task')
        custom_task = dag.task_dict['custom_task']
        
        assert custom_task.operator_extra_links
        session = settings.Session()
        self.scheduler_job = SchedulerJob(executor=self.null_exec)
        self.scheduler_job.processor_agent = mock.MagicMock()
        self.scheduler_job._start_queued_dagruns(session)
        session.flush()
        
        s_dag_2 = self.scheduler_job.dagbag.get_dag(dag.dag_id)
        custom_task = s_dag_2.task_dict['custom_task']
        
        assert not custom_task.operator_extra_links
    def test_scheduler_create_dag_runs_does_not_raise_error(self, caplog, dag_maker):