**Apache Airflow version**: 2.1.3
**OS**: Custom Docker Image built from python:3.8-slim
**Apache Airflow Provider versions**:
apache-airflow-providers-cncf-kubernetes==2.0.1
apache-airflow-providers-ftp==2.0.0
apache-airflow-providers-http==2.0.0
apache-airflow-providers-imap==2.0.0

**Deployment**: Self managed on EKS, manifests templated and rendered using krane, dags mounted via PVC & EFS

**What happened**: Task remains in queued state 
<img width="1439" alt="Screen Shot 2021-08-18 at 12 17 28 PM" src="https://user-images.githubusercontent.com/47788186/129936170-e16e1362-24ca-4ce9-b2f7-978f2642d388.png">

**What you expected to happen**: Task starts running

**How to reproduce it**: I believe it is because a node is removed. I've attached both the scheduler/k8s executor logs and the kubernetes logs.  I deployed a custom executor with an extra line in the KubernetesJobWatcher.process_status that logged the Event type for the pod.


```
date,name,message
2021-09-08T01:06:35.083Z,KubernetesJobWatcher,Event: <pod_id> had an event of type ADDED
2021-09-08T01:06:35.084Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:35.085Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:35.085Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:43.390Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:43.390Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:43.390Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:43.391Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:45.392Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:45.393Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running
2021-09-08T01:07:33.185Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:07:33.186Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running
2021-09-08T01:09:50.478Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:09:50.479Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running
2021-09-08T01:09:50.479Z,KubernetesJobWatcher,Event: <pod_id> had an event of type DELETED
2021-09-08T01:09:50.480Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running

```

Here are the corresponding EKS Logs

```
@timestamp,@message
2021-09-08 01:06:34      1 factory.go:503] pod etl/<pod_id> is already present in the backoff queue
2021-09-08 01:06:43      1 scheduler.go:742] pod etl/<pod_id> is bound successfully on node ""ip-10-0-133-132.ec2.internal"", 19 nodes evaluated, 1 nodes were found feasible."
2021-09-08 01:07:32      1 controller_utils.go:122] Update ready status of pods on node [ip-10-0-133-132.ec2.internal]
2021-09-08 01:07:32      1 controller_utils.go:139] Updating ready status of pod <pod_id> to false
2021-09-08 01:07:38      1 node_lifecycle_controller.go:889] Node ip-10-0-133-132.ec2.internal is unresponsive as of 2021-09-08 01:07:38.017788167
2021-09-08 01:08:51      1 node_tree.go:100] Removed node "ip-10-0-133-132.ec2.internal" in group "us-east-1:\x00:us-east-1b" from NodeTree
2021-09-08 01:08:53      1 node_lifecycle_controller.go:789] Controller observed a Node deletion: ip-10-0-133-132.ec2.internal
2021-09-08 01:09:50      1 gc_controller.go:185] Found orphaned Pod etl/<pod_id> assigned to the Node ip-10-0-133-132.ec2.internal. Deleting.
2021-09-08 01:09:50      1 gc_controller.go:189] Forced deletion of orphaned Pod etl/<pod_id> succeeded

```
**Are you willing to submit a PR?** I think there is a state & event combination missing from the process_state function in the KubernetesJobWatcher. I will submit a PR to fix.

airflow/executors/kubernetes_executor.py
"""Executor for Kubernetes"""
    def __init__(self):
        self.kube_config = KubeConfig()
        self._manager = multiprocessing.Manager()
        self.task_queue: 'Queue[KubernetesJobType]' = self._manager.Queue()
        self.result_queue: 'Queue[KubernetesResultsType]' = self._manager.Queue()
        self.kube_scheduler: Optional[AirflowKubernetesScheduler] = None
        self.kube_client: Optional[client.CoreV1Api] = None
        self.scheduler_job_id: Optional[str] = None
        self.event_scheduler: Optional[EventScheduler] = None
        super().__init__(parallelism=self.kube_config.parallelism)
    @provide_session
    def clear_not_launched_queued_tasks(self, session=None) -> None:
        """
        If the airflow scheduler restarts with pending "Queued" tasks, the tasks may or
        may not
        have been launched. Thus on starting up the scheduler let's check every
        "Queued" task to
        see if it has been launched (ie: if there is a corresponding pod on kubernetes)
        If it has been launched then do nothing, otherwise reset the state to "None" so
        the task
        will be rescheduled
        This will not be necessary in a future version of airflow in which there is
        proper support
        for State.LAUNCHED
        """
        self.log.debug("Clearing tasks that have not been launched")
        if not self.kube_client:
            raise AirflowException(NOT_STARTED_MESSAGE)
        queued_tasks = session.query(TaskInstance).filter(TaskInstance.state == State.QUEUED).all()