
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 scheduler._create_dag_runs does not raise an error when the DAG does not exist
        in serialized_dag table
        """
        with dag_maker(dag_id='test_scheduler_create_dag_runs_does_not_raise_error', serialized=False):
            DummyOperator(
                task_id='dummy',
            )
        self.scheduler_job = SchedulerJob(subdir=os.devnull, executor=self.null_exec)
        self.scheduler_job.processor_agent = mock.MagicMock()
        caplog.set_level('FATAL')
        caplog.clear()
        with create_session() as session, caplog.at_level(
            'ERROR',
            logger='airflow.jobs.scheduler_job',
        ):
            self.scheduler_job._create_dag_runs([dag_maker.dag_model], session)
            assert caplog.messages == [
                "DAG 'test_scheduler_create_dag_runs_does_not_raise_error' not found in serialized_dag table",
            ]
    def test_bulk_write_to_db_external_trigger_dont_skip_scheduled_run(self, dag_maker):