airflow: Kubernetes executor is broken in Airflow 1.10.11

Apache Airflow version: 1.10.11

Kubernetes version (if you are using kubernetes) (use kubectl version): v1.15.11

Environment:

• Cloud provider or hardware configuration: AWS
• OS (e.g. from /etc/os-release):
• Kernel (e.g. uname -a):
• Install tools:
• Others:

What happened:

The scheduler can’t launch pods if KubernetesExecutor defines pod resources The error message:

[2020-07-14 08:00:13,853] {{kubernetes_executor.py:758}} INFO - Add task ('DAG_ID', 'TASK_ID', datetime.datetime(2020, 7, 14, 7, 0, tzinfo=<Timezone [UTC]>), 1) with command ['airflow', 'run', 'DAG_ID', 'TASK_ID', '2020-07-14T07:00:00+00:00', '--local', '--pool', 'rider_scoring', '-sd', '/opt/airflow/dags/dag.py'] with executor_config {'KubernetesExecutor': {'request_cpu': '200m', 'limit_cpu': '200m', 'request_memory': '500Mi', 'limit_memory': '500Mi'}}
[2020-07-14 08:00:13,854] {{scheduler_job.py:1383}} ERROR - Exception when executing execute_helper
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1381, in _execute
    self._execute_helper()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1452, in _execute_helper
    if not self._validate_and_run_task_instances(simple_dag_bag=simple_dag_bag):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1514, in _validate_and_run_task_instances
    self.executor.heartbeat()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 130, in heartbeat
    self.trigger_tasks(open_slots)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 154, in trigger_tasks
    executor_config=simple_ti.executor_config)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/kubernetes_executor.py", line 761, in execute_async
    kube_executor_config = PodGenerator.from_obj(executor_config)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/kubernetes/pod_generator.py", line 374, in from_obj
    return PodGenerator(**namespaced).gen_pod()
TypeError: __init__() got an unexpected keyword argument 'request_cpu'

What you expected to happen:

The scheduler should launch new worker pods

How to reproduce it:

Define pod resources in KubernetesExecutor

default_args = {
    'executor_config': {
        'KubernetesExecutor': {
            'request_cpu': "200m",
            'limit_cpu': "200m",
            'request_memory': "500Mi",
            'limit_memory': "500Mi"
        }
    }
}
dag = DAG(
    dag_id='foo',
    default_args=default_args,
)

Anything else we need to know:

I think there are two issues:

• When #6230 has been backported to 1.10.11 branch, namespaced['resources'] = resources get lost. Commit in 1.10.11
• PodGenerator gets as parameters namespaced object which contains the content of KubernetesExecutor. When resources are defined in KubernetesExecutor, PodGenerator receives request_cpu, limit_cpu, request_memory, limit_memory but they are not valid parameters for PodGenerator. I believe we need to delete them from namespaced. Another option would be to instantiate namespaced as an empty dict and add only the allowed parameter for PodGenerator.

FYI @dimberman @kaxil