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
)
        
        self.scheduler_job.processor_agent.send_callback_to_execute.assert_called_once_with(expected_callback)
        
        
        
        real_dag = self.scheduler_job.dagbag.get_dag(dag.dag_id)
        self.scheduler_job._send_sla_callbacks_to_processor.assert_called_once_with(real_dag)
        session.rollback()
        session.close()
    def test_dagrun_callbacks_commited_before_sent(self, dag_maker):
        """
        Tests that before any callbacks are sent to the processor, the session is committed. This ensures
        that the dagrun details are up to date when the callbacks are run.
        """
        with dag_maker(dag_id='test_dagrun_callbacks_commited_before_sent'):
            DummyOperator(task_id='dummy')
        self.scheduler_job = SchedulerJob(subdir=os.devnull)
        self.scheduler_job.processor_agent = mock.Mock()
        self.scheduler_job._send_dag_callbacks_to_processor = mock.Mock()
        self.scheduler_job._schedule_dag_run = mock.Mock()
        dr = dag_maker.create_dagrun()
        session = settings.Session()
        ti = dr.get_task_instance('dummy')
        ti.set_state(State.SUCCESS, session)
        with mock.patch.object(settings, "USE_JOB_SCHEDULE", False), mock.patch(
            "airflow.jobs.scheduler_job.prohibit_commit"
        ) as mock_gaurd:
            mock_gaurd.return_value.__enter__.return_value.commit.side_effect = session.commit
            def mock_schedule_dag_run(*args, **kwargs):