**Apache Airflow version**: 2.0.0rc2
**Environment**: Breeze with example dags, Python 3.8 postgres.
- **OS** (e.g. from /etc/os-release): Linux
- **Kernel** (e.g. `uname -a`): Breeze CI image
- **Install tools**: Breeze: 
- **Executor**: LocalExecutor
```
./breeze start-airflow --backend postgres --load-example-dags --load-default-connections --install-airflow-version 2.0.0rc2 --skip-mounting-local-sources --python 3.8
```

**What happened**:

When testing airflow logging I occasionally stumble upon "KeyError' from `dag_procesing.py`. I am not sure exactly when it happens. It's not always reproducible but it looks like it is when I restart scheduler and trigger 'example_bash_operator.py" it happens rather randomly (1/10 times more or less). It does not happen always when I triggere task manually. DAG gets correctly executed after triggering, but the log is there and warniing printed in the logs right after the DAG finishes execution.

The error I see in scheduler's logs:

```
[2020-12-13 19:35:33,752] {dagbag.py:440} INFO - Filling up the DagBag from /usr/local/lib/python3.8/site-packages/airflow/example_dags/example_bash_operator.py                                                                                                                                          
Running <TaskInstance: example_bash_operator.run_after_loop 2020-12-13T19:35:30.648020+00:00 [queued]> on host 6611da4b1a27
[2020-12-13 19:35:34,517] {dagrun.py:444} INFO - Marking run <DagRun example_bash_operator @ 2020-12-13 19:35:30.648020+00:00: manual__2020-12-13T19:35:30.648020+00:00, externally triggered: True> successful                                                                                           
[2020-12-13 19:35:34,523] {scheduler_job.py:1193} INFO - Executor reports execution of example_bash_operator.run_after_loop execution_date=2020-12-13 19:35:30.648020+00:00 exited with status success for try_number 1                                                                                   
Process ForkProcess-34:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/dag_processing.py", line 365, in _run_processor_manager
    processor_manager.start()
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/dag_processing.py", line 596, in start
    return self._run_parsing_loop()
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/dag_processing.py", line 659, in _run_parsing_loop
    self._processors.pop(processor.file_path)
KeyError: '/usr/local/lib/python3.8/site-packages/airflow/example_dags/example_bash_operator.py'
[2020-12-13 19:35:35,589] {dag_processing.py:396} WARNING - DagFileProcessorManager (PID=1029759) exited with exit code 1 - re-launching
```

**What you expected to happen**:

No error in logs.

**How to reproduce it**:

```
./breeze start-airflow --backend postgres --load-example-dags --load-default-connections --install-airflow-version 2.0.0rc2 --skip-mounting-local-sources --python 3.8
```

Login to the webserver, enable 'example_bash_operator", wait for it to execute. Trigger the example DAG several times (always wait for the end of execution. It happens randomly (for me around 1/10 tasks)


**Anything else we need to know**:

airflow/utils/dag_processing.py
"""
        Use multiple processes to parse and generate tasks for the
        DAGs in parallel. By processing them in separate processes,
        we can get parallelism and isolation from potentially harmful
        user code.
        """
        self.register_exit_signals()
        
        os.setpgid(0, 0)
        self.log.info("Processing files using up to %s processes at a time ", self._parallelism)
        self.log.info("Process each file at most once every %s seconds", self._file_process_interval)
        self.log.info(
            "Checking for new files in %s every %s seconds", self._dag_directory, self.dag_dir_list_interval
        )
        return self._run_parsing_loop()
    def _run_parsing_loop(self):
        
        if self._async_mode:
            poll_time = 0.0
        else:
            poll_time = None
        self._refresh_dag_dir()
        self.prepare_file_path_queue()