**Apache Airflow version**: 2.0.1


**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): 1.15

**Environment**:

- **Cloud provider or hardware configuration**: k8s on bare metal
- **OS** (e.g. from /etc/os-release): 
- **Kernel** (e.g. `uname -a`):
- **Install tools**: pip3
- **Others**:

**What happened**:
Scheduler dies with 
```
[2021-02-10 21:09:27,469] {scheduler_job.py:1298} ERROR - Exception when executing SchedulerJob._run_schedu
ler_loop
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1280, in _execute
    self._run_scheduler_loop()
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1384, in _run_scheduler
_loop
    self.executor.heartbeat()
  File "/usr/local/lib/python3.8/site-packages/airflow/executors/base_executor.py", line 158, in heartbeat
    self.trigger_tasks(open_slots)
  File "/usr/local/lib/python3.8/site-packages/airflow/executors/base_executor.py", line 188, in trigger_ta
sks
    self.execute_async(key=key, command=command, queue=None, executor_config=ti.executor_config)
  File "/usr/local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 493, in exec
ute_async
    kube_executor_config = PodGenerator.from_obj(executor_config)
  File "/usr/local/lib/python3.8/site-packages/airflow/kubernetes/pod_generator.py", line 175, in from_obj
    k8s_legacy_object = obj.get("KubernetesExecutor", None)
AttributeError: 'V1Pod' object has no attribute 'get'
[2021-02-10 21:09:28,475] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 60
[2021-02-10 21:09:29,222] {process_utils.py:66} INFO - Process psutil.Process(pid=66, status='terminated',
started='21:09:27') (66) terminated with exit code None
[2021-02-10 21:09:29,697] {process_utils.py:206} INFO - Waiting up to 5 seconds for processes to exit...
[2021-02-10 21:09:29,716] {process_utils.py:66} INFO - Process psutil.Process(pid=75, status='terminated',
started='21:09:28') (75) terminated with exit code None
[2021-02-10 21:09:29,717] {process_utils.py:66} INFO - Process psutil.Process(pid=60, status='terminated',
exitcode=0, started='21:09:27') (60) terminated with exit code 0
[2021-02-10 21:09:29,717] {scheduler_job.py:1301} INFO - Exited execute loop
```

**What you expected to happen**:
DAG loading fails, producing an error for just that DAG, instead of crashing the scheduler.

**How to reproduce it**:
Create a task like
```
    test = DummyOperator(task_id="new-pod-spec",
                              executor_config=k8s.V1Pod(
                                      spec=k8s.V1PodSpec(
                                          containers=[
                                              k8s.V1Container(
                                                  name="base",
                                                  image="myimage",
                                                  image_pull_policy="Always"
                                              )
                                          ]
                                      )))
```
or 
```
    test = DummyOperator(task_id="new-pod-spec",
                              executor_config={"KubernetesExecutor": k8s.V1Pod(
                                      spec=k8s.V1PodSpec(
                                          containers=[
                                              k8s.V1Container(
                                                  name="base",
                                                  image="myimage",
                                                  image_pull_policy="Always"
                                              )
                                          ]
                                      ))})
```
essentially anything where it expects a dict but gets something else, and run the scheduler using the kubernetes executor

tests/executors/test_kubernetes_executor.py
executor = self.kubernetes_executor
        executor.scheduler_job_id = "modified"
        pod_ids = {"dagtask": {}}
        pod = k8s.V1Pod(
            metadata=k8s.V1ObjectMeta(
                name="foo", labels={"airflow-worker": "bar", "dag_id": "dag", "task_id": "task"}
            )
        )
        executor.adopt_launched_task(mock_kube_client, pod=pod, pod_ids=pod_ids)
        assert mock_kube_client.patch_namespaced_pod.call_args[1] == {
            'body': {
                'metadata': {
                    'labels': {'airflow-worker': 'modified', 'dag_id': 'dag', 'task_id': 'task'},
                    'name': 'foo',
                }
            },
            'name': 'foo',
            'namespace': None,
        }
        assert pod_ids == {}
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    def test_not_adopt_unassigned_task(self, mock_kube_client):