**Apache Airflow version**: 2.0.1
**Kubernetes version**: v1.19.7
**Executor**: KubernetesExecutor

**What happened**:

If you have a worker that gets stuck in pending forever, say with a missing volume mount, the task will stay in the queued state forever. Nothing is applying a timeout on it actually being able to start.

**What you expected to happen**:

Eventually the scheduler will notice that the worker hasn't progressed past pending after a given amount of time and will mark it as a failure.

**How to reproduce it**:
```python
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from kubernetes.client import models as k8s


default_args = {
    "owner": "airflow",
}

with DAG(
    dag_id="pending",
    default_args=default_args,
    schedule_interval=None,
    start_date=days_ago(2),
) as dag:
    BashOperator(
        task_id="forever_pending",
        bash_command="date; sleep 30; date",
        executor_config={
            "pod_override": k8s.V1Pod(
                spec=k8s.V1PodSpec(
                    containers=[
                        k8s.V1Container(
                            name="base",
                            volume_mounts=[
                                k8s.V1VolumeMount(mount_path="/foo/", name="vol")
                            ],
                        )
                    ],
                    volumes=[
                        k8s.V1Volume(
                            name="vol",
                            persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(
                                claim_name="missing"
                            ),
                        )
                    ],
                )
            ),
        },
    )
```


**Anything else we need to know**:
Related to:

* #15149 (This is reporting that these pending pods don't get deleted via "Mark Failed")
* #14556 (This handles when these pending pods get deleted and is already fixed)

airflow/jobs/scheduler_job.py
super().__init__()
        self.dag_ids = dag_ids
        self._log = log
    @provide_session
    def manage_slas(self, dag: DAG, session: Session = None) -> None:
        """
        Finding all tasks that have SLAs defined, and sending alert emails
        where needed. New SLA misses are also recorded in the database.
        We are assuming that the scheduler runs often, so we only check for
        tasks that should have succeeded in the past hour.
        """
        self.log.info("Running SLA Checks for %s", dag.dag_id)
        if not any(isinstance(ti.sla, timedelta) for ti in dag.tasks):
            self.log.info("Skipping SLA check for %s because no tasks in DAG have SLAs", dag)
            return
        qry = (
            session.query(TI.task_id, func.max(TI.execution_date).label('max_ti'))
            .with_hint(TI, 'USE INDEX (PRIMARY)', dialect_name='mysql')
            .filter(TI.dag_id == dag.dag_id)
            .filter(or_(TI.state == State.SUCCESS, TI.state == State.SKIPPED))
            .filter(TI.task_id.in_(dag.task_ids))
            .group_by(TI.task_id)
            .subquery('sq')
        )
        max_tis: List[TI] = (
            session.query(TI)
            .filter(
                TI.dag_id == dag.dag_id,
                TI.task_id == qry.c.task_id,
                TI.execution_date == qry.c.max_ti,