From 7813076ac76a0bea858658cd30b224341c637a0b Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Wed, 14 Jul 2021 17:33:24 +0100
Subject: [PATCH] [backport] Fix bug with `executor_config` and Volumes

Without this fix the following DAG will not create Volumes & Volume Mounts and fais:

```python
dag1 = DAG(
    dag_id='example_k8s-dags',
    schedule_interval='@once',
    start_date=datetime(2020, 1, 1)
)
p = PythonOperator(
    task_id='test',
    dag=dag1,
    python_callable=lambda: 1,
    executor_config={
        "KubernetesExecutor": {
            'resources': {
                'limits': {'memory': '200Mi', 'cpu': '100m'},
                'requests': {'memory': '100Mi', 'cpu': '100m'}
            },
            "volumes": [
                {
                    "name": "tmp-volume",
                    "hostPath": {"path": "/tmp/"},
                },
            ],
            "volume_mounts": [
                {
                    "mountPath": "/tmp/",
                    "name": "tmp-volume",
                },
            ],
        }
    }
)
```
---
 airflow/kubernetes/pod_generator.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/airflow/kubernetes/pod_generator.py b/airflow/kubernetes/pod_generator.py
index 211b0275ff59..cca862653ff8 100644
--- a/airflow/kubernetes/pod_generator.py
+++ b/airflow/kubernetes/pod_generator.py
@@ -32,7 +32,7 @@
 import yaml
 from dateutil import parser
 from kubernetes.client.api_client import ApiClient
-from airflow.contrib.kubernetes.pod import _extract_volume_mounts
+from airflow.contrib.kubernetes.pod import _extract_volume_mounts, _extract_volumes
 
 from airflow.exceptions import AirflowConfigException
 from airflow.version import version as airflow_version
@@ -281,7 +281,7 @@ def __init__(
         self.spec.affinity = affinity
         self.spec.service_account_name = service_account_name
         self.spec.init_containers = init_containers
-        self.spec.volumes = volumes or []
+        self.spec.volumes = [v.to_k8s_client_obj() for v in _extract_volumes(volumes)] or []
         self.spec.node_selector = node_selectors
         self.spec.restart_policy = restart_policy
         self.spec.priority_class_name = priority_class_name