The test fails occasionally with the below stacktrace, so I am marking this as Quarantined.

```
  _______________ TestSchedulerJob.test_scheduler_verify_pool_full _______________
  
  self = <tests.jobs.test_scheduler_job.TestSchedulerJob object at 0x7fbaaaba0f40>
  
      def test_scheduler_verify_pool_full(self):
          """
          Test task instances not queued when pool is full
          """
          dag = DAG(dag_id='test_scheduler_verify_pool_full', start_date=DEFAULT_DATE)
      
          BashOperator(
              task_id='dummy',
              dag=dag,
              owner='airflow',
              pool='test_scheduler_verify_pool_full',
              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, root_dag=dag)
          dagbag.sync_to_db()
      
          session = settings.Session()
          pool = Pool(pool='test_scheduler_verify_pool_full', slots=1)
          session.add(pool)
          session.flush()
      
          dag = SerializedDAG.from_dict(SerializedDAG.to_dict(dag))
          SerializedDagModel.write_dag(dag)
      
          self.scheduler_job = SchedulerJob(executor=self.null_exec)
          self.scheduler_job.processor_agent = mock.MagicMock()
      
          # Create 2 dagruns, which will create 2 task instances.
          dr = dag.create_dagrun(
              run_type=DagRunType.SCHEDULED,
              execution_date=DEFAULT_DATE,
              state=State.RUNNING,
          )
  >       self.scheduler_job._schedule_dag_run(dr, session)
  
  tests/jobs/test_scheduler_job.py:2108: 
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  airflow/jobs/scheduler_job.py:1020: in _schedule_dag_run
      dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
  airflow/utils/session.py:67: in wrapper
      return func(*args, **kwargs)
  airflow/models/dagbag.py:186: in get_dag
      self._add_dag_from_db(dag_id=dag_id, session=session)
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  
  self = <airflow.models.dagbag.DagBag object at 0x7fbaaab95100>
  dag_id = 'test_scheduler_verify_pool_full'
  session = <sqlalchemy.orm.session.Session object at 0x7fbaaad9d0a0>
  
      def _add_dag_from_db(self, dag_id: str, session: Session):
          """Add DAG to DagBag from DB"""
          from airflow.models.serialized_dag import SerializedDagModel
      
          row = SerializedDagModel.get(dag_id, session)
          if not row:
  >           raise SerializedDagNotFound(f"DAG '{dag_id}' not found in serialized_dag table")
  E           airflow.exceptions.SerializedDagNotFound: DAG 'test_scheduler_verify_pool_full' not found in serialized_dag table
```
tests/jobs/test_scheduler_job.py
scheduler.processor_agent = mock.MagicMock()
        def _create_dagruns(dag: DAG):
            next_info = dag.next_dagrun_info(None)
            assert next_info is not None
            for _ in range(5):
                yield dag.create_dagrun(
                    run_type=DagRunType.SCHEDULED,
                    execution_date=next_info.logical_date,
                    data_interval=next_info.data_interval,
                    state=State.RUNNING,
                )
                next_info = dag.next_dagrun_info(next_info.data_interval)
                if next_info is None:
                    break
        
        
        
        for dr in _create_dagruns(dag_d1):
            scheduler._schedule_dag_run(dr, session)
        for dr in _create_dagruns(dag_d2):
            scheduler._schedule_dag_run(dr, session)
        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)
        
        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
        )
    def test_scheduler_verify_priority_and_slots(self, dag_maker):