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
"""Ensure orphaned tasks with non-running dagruns are not reset."""
        dag_id = 'test_reset_orphaned_tasks_non_running_dagruns'
        with dag_maker(dag_id=dag_id, schedule_interval='@daily'):
            task_id = dag_id + '_task'
            DummyOperator(task_id=task_id)
        self.scheduler_job = SchedulerJob(subdir=os.devnull)
        session = settings.Session()
        session.add(self.scheduler_job)
        session.flush()
        dr1 = dag_maker.create_dagrun()
        tis = dr1.get_task_instances(session=session)
        assert 1 == len(tis)
        tis[0].state = State.SCHEDULED
        tis[0].queued_by_job_id = self.scheduler_job.id
        session.merge(dr1)
        session.merge(tis[0])
        session.flush()
        assert 0 == self.scheduler_job.adopt_or_reset_orphaned_tasks(session=session)
        session.rollback()
    def test_adopt_or_reset_orphaned_tasks_stale_scheduler_jobs(self, dag_maker):
        dag_id = 'test_adopt_or_reset_orphaned_tasks_stale_scheduler_jobs'
        with dag_maker(dag_id=dag_id, schedule_interval='@daily'):
            DummyOperator(task_id='task1')
            DummyOperator(task_id='task2')
        self.scheduler_job = SchedulerJob(subdir=os.devnull)
        session = settings.Session()
        self.scheduler_job.state = State.RUNNING
        self.scheduler_job.latest_heartbeat = timezone.utcnow()
        session.add(self.scheduler_job)