**Apache Airflow version**:

1.10.12

**What happened**:

I created dag with dagrun_timeout=2 minutes. 

After 2 minutes dagrun is marked as failed and the next one is started, but task keeps going.

**What you expected to happen**:

Task is killed with dag run as it is done when you mark dagrun failed manually.

**How to reproduce it**:
```
dag = DAG(dag_id='platform.airflow-test',
          description='',
          schedule_interval="0 0 * * *",
          start_date=datetime(2020, 7, 1),
          max_active_runs=1,
          catchup=True,
          dagrun_timeout=timedelta(minutes=2))

run_this = BashOperator(
    task_id='run_after_loop',
    bash_command=' for((i=1;i<=600;i+=1)); do echo "Welcome $i times"; sleep 1;  done',
    dag=dag,
)
```
airflow/jobs/scheduler_job.py
self.log.exception("Could not send SLA Miss email notification for DAG %s", dag.dag_id)
            
            if notification_sent:
                for sla in slas:
                    sla.email_sent = email_sent
                    sla.notification_sent = True
                    session.merge(sla)
            session.commit()
    @staticmethod
    def update_import_errors(session: Session, dagbag: DagBag) -> None:
        """
        For the DAGs in the given DagBag, record any associated import errors and clears
        errors for files that no longer have them. These are usually displayed through the
        Airflow UI so that users know that there are issues parsing DAGs.
        :param session: session for ORM operations
        :type session: sqlalchemy.orm.session.Session
        :param dagbag: DagBag containing DAGs with import errors
        :type dagbag: airflow.DagBag
        """
        
        for dagbag_file in dagbag.file_last_changed:
            session.query(errors.ImportError).filter(errors.ImportError.filename == dagbag_file).delete()
        
        for filename, stacktrace in dagbag.import_errors.items():
            session.add(
                errors.ImportError(filename=filename, timestamp=timezone.utcnow(), stacktrace=stacktrace)
            )
        session.commit()
    @provide_session
    def execute_callbacks(