**Apache Airflow version**: 2.0.0a2

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): v1.15.12

**Environment**:
- **OS** (e.g. from /etc/os-release): CentOS 7


**What happened**:

When using KubernetesExecutor, a single SIGTERM doesn't result in the scheduler actually exiting cleanly. However, it will exit after receiving a second SIGTERM.

The same behavior happens if you have an exception in the scheduler. An easy way to replicate it is to raise in `pod_mutation_hook`.

In practice, this means the scheduler doesn't get restarted by K8s after encountering a exception. Or, when terminating the scheduler pod, it sticks around until k8s ultimately sends it a SIGKILL after its grace period is up.

**What you expected to happen**:

The scheduler to exit after a single SIGTERM.

**How to reproduce it**:
The easiest way to reproduce it is to simply send a SIGKILL to your KubernetesExecutor scheduler pod:

```
$ kubectl exec {scheduler_pod_name} kill 1
```

Note, the pod stays running, while the logs report:

```
[2020-10-30 17:34:09,581] {scheduler_job.py:794} INFO - Exiting gracefully upon receiving signal 15
[2020-10-30 17:34:10,586] {process_utils.py:97} INFO - Sending Signals.SIGTERM to GPID 59
[2020-10-30 17:34:10,879] {process_utils.py:63} INFO - Process psutil.Process(pid=59, status='terminated', exitcode=0, started=
'17:33:23') (59) terminated with exit code 0
[2020-10-30 17:34:10,882] {process_utils.py:97} INFO - Sending Signals.SIGTERM to GPID 59
[2020-10-30 17:34:10,882] {scheduler_job.py:1329} INFO - Exited execute loop
```

Issue another `kill`:
```
$ kubectl exec {scheduler_pod_name} kill 1
```

And the pod will be restarted by k8s.

If you can't reproduce it, let me know and I'll try and get something running on kind.

**Anything else we need to know**:

The same behavior happens if you have an exception in the scheduler. An easy way to replicate it is to raise in a `pod_mutation_hook`.


airflow/jobs/scheduler_job.py
self,
        dag_run: DagRun,
        currently_active_runs: Set[datetime.datetime],
        session: Session,
    ) -> int:
        """
        Make scheduling decisions about an individual dag run
        ``currently_active_runs`` is passed in so that a batch query can be
        used to ask this for all dag runs in the batch, to avoid an n+1 query.
        :param dag_run: The DagRun to schedule
        :param currently_active_runs: Number of currently active runs of this DAG
        :return: Number of tasks scheduled
        """
        dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
        if not dag:
            self.log.error("Couldn't find dag %s in DagBag/DB!", dag_run.dag_id)
            return 0
        if (
            dag_run.start_date
            and dag.dagrun_timeout
            and dag_run.start_date < timezone.utcnow() - dag.dagrun_timeout
        ):
            dag_run.set_state(State.FAILED)