**Apache Airflow version**: 2.0.0, LocalExecutor

**Environment**: Docker on Win10 with WSL, official Python3.8 image

**What happened**:

When a DAG is running, and I delete task from the running DAG, the scheduler fails. When using Docker, upon automatic restart of the scheduler, the scheduler just fails again, perpetually.

![image](https://user-images.githubusercontent.com/46958547/103558919-a89f6e80-4eb5-11eb-8521-19b04ee3c690.png)

Note: I don't _know_ if the task itself was running at the time, but I would guess it was.

**What you expected to happen**:

The scheduler should understand that the task is not part of the DAG anymore and not fail.

**How to reproduce it**:

- Create a DAG with multiple tasks
- Let it run
- While running, delete one of the tasks from the source code
- See the scheduler break
airflow/models/dagrun.py
The emitted data may contains outlier (e.g. when the first task was cleared, so
        the second task's start_date will be used), but we can get rid of the the outliers
        on the stats side through the dashboards tooling built.
        Note, the stat will only be emitted if the DagRun is a scheduler triggered one
        (i.e. external_trigger is False).
        """
        if self.state == State.RUNNING:
            return
        if self.external_trigger:
            return
        if not finished_tis:
            return
        try:
            dag = self.get_dag()
            if not self.dag.schedule_interval or self.dag.schedule_interval == "@once":
                
                return
            ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
            ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
            first_start_date = ordered_tis_by_start_date[0].start_date
            if first_start_date:
                
                
                
                true_delay = first_start_date - dag.following_schedule(self.execution_date)
                if true_delay.total_seconds() > 0:
                    Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)
        except Exception as e:
            self.log.warning(f'Failed to record first_task_scheduling_delay metric:\n{e}')
    def _emit_duration_stats_for_finished_state(self):