The `TestSchedulerJob.test_retry_still_in_executor` fails occasionally and should be quarantined.

```
  ________________ TestSchedulerJob.test_retry_still_in_executor _________________
  
  self = <tests.jobs.test_scheduler_job.TestSchedulerJob object at 0x7f4c9031f128>
  
      def test_retry_still_in_executor(self):
          """
          Checks if the scheduler does not put a task in limbo, when a task is retried
          but is still present in the executor.
          """
          executor = MockExecutor(do_update=False)
          dagbag = DagBag(dag_folder=os.path.join(settings.DAGS_FOLDER, "no_dags.py"), include_examples=False)
          dagbag.dags.clear()
      
          dag = DAG(dag_id='test_retry_still_in_executor', start_date=DEFAULT_DATE, schedule_interval="@once")
          dag_task1 = BashOperator(
              task_id='test_retry_handling_op', bash_command='exit 1', retries=1, dag=dag, owner='airflow'
          )
      
          dag.clear()
          dag.is_subdag = False
      
          with create_session() as session:
              orm_dag = DagModel(dag_id=dag.dag_id)
              orm_dag.is_paused = False
              session.merge(orm_dag)
      
          dagbag.bag_dag(dag=dag, root_dag=dag)
          dagbag.sync_to_db()
      
          @mock.patch('airflow.dag_processing.processor.DagBag', return_value=dagbag)
          def do_schedule(mock_dagbag):
              # Use a empty file since the above mock will return the
              # expected DAGs. Also specify only a single file so that it doesn't
              # try to schedule the above DAG repeatedly.
              self.scheduler_job = SchedulerJob(
                  num_runs=1, executor=executor, subdir=os.path.join(settings.DAGS_FOLDER, "no_dags.py")
              )
              self.scheduler_job.heartrate = 0
              self.scheduler_job.run()
      
          do_schedule()
          with create_session() as session:
              ti = (
                  session.query(TaskInstance)
                  .filter(
                      TaskInstance.dag_id == 'test_retry_still_in_executor',
                      TaskInstance.task_id == 'test_retry_handling_op',
                  )
                  .first()
              )
  >       ti.task = dag_task1
  E       AttributeError: 'NoneType' object has no attribute 'task'
  
  tests/jobs/test_scheduler_job.py:2514: AttributeError
```
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):