<p>I am using airflow 2.4.3 and running KubernetesPodOperator</p>
<p>Below is the code and error:-</p>
<p>Please help me with creating a KubernetesPosOperator in python. I have tried on both GCP and Azure.</p>
<p>Also adding the kubernetes documentation for reference:-</p>
<p><a href="https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes/stable/_api/airflow/providers/cncf/kubernetes/operators/kubernetes_pod/index.html#airflow.providers.cncf.kubernetes.operators.kubernetes_pod.KubernetesPodOperator" rel="nofollow noreferrer">https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes/stable/_api/airflow/providers/cncf/kubernetes/operators/kubernetes_pod/index.html#airflow.providers.cncf.kubernetes.operators.kubernetes_pod.KubernetesPodOperator</a></p>
<p>I can also share any other info if required.</p>
<pre><code>
from kubernetes.client import models as k8s

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

#custom modules
from spark_API.spark_submit import SparkSubmit

#import modules
import json
import datetime
import logging

import airflow_API
import time


airflow_dag_object = AirflowDagUtilities(&quot;aaa_test_airflow_api&quot;)


def def_func1(**kwargs):
    print(&quot;In func1&quot;)
    
    namespace = &quot;segmentation-pipeline&quot;
    docker_image = &quot;****:v6&quot; # name commented out
    is_delete_operator_pod = True
    
    docker_image_creds =  [k8s.V1LocalObjectReference(&quot;****&quot;)] # harbor name commented out

    
    submit_command = [&quot;/bin/bash&quot;,&quot;-c&quot;]
    
    max_cores = 60
    driver_memory = &quot;4g&quot;
    executor_memory = &quot;4g&quot;
            
    submit_args = &quot;/usr/local/spark/bin/spark-submit --master local[&quot; + str(max_cores) + &quot;] --driver-memory &quot; + \
                                driver_memory + &quot; --executor-memory &quot; + executor_memory + &quot; &quot;
    
    submit_spark_pipeline_config_conf = &quot;--conf &quot; + '\'' + 'spark.pipelineConfig' + &quot;=&quot; + json.dumps(_infra_config.get_infra_config(),separators=(',',':')) + '\'' + &quot; &quot;
    
    submit_spark_broadcast_timeout = &quot;--conf &quot; + '\&quot;' + &quot;spark.sql.broadcastTimeout&quot; + &quot;=&quot; + str(&quot;36000&quot;) + '\&quot;' + &quot; &quot;
    
    submit_spark_max_result_size = &quot;--conf &quot; + '\&quot;' + &quot;spark.driver.maxResultSize&quot; + &quot;=&quot; + str(&quot;0&quot;) + '\&quot;' + &quot; &quot;

    final_dependency_jars = [&quot;./resources/mysql_connector_java_5.1.45.jar&quot;,\
                            &quot;./resources/commons_httpclient_3.0.1.jar&quot;]
                 
    
    dependency_jars_string = ','.join(list(set(final_dependency_jars)))
    
    submit_spark_dependency_jars = &quot;--conf &quot; + '\&quot;' + &quot;spark.jars&quot; + &quot;=&quot; + dependency_jars_string + '\&quot;' + &quot; &quot;
    extra_conf = []
    extra_conf_final = []
    
    for conf in extra_conf:
        conf_appended_string = &quot;--conf &quot; + '\&quot;' + conf + '\'' + &quot; &quot;
        extra_conf_final.append(conf_appended_string)
            
    
    extra_conf = &quot; &quot;.join(extra_conf_final) + &quot; &quot;
    
    airflow_task_settings = airflow_API.extract_airflow_task_details(kwargs['task_instance'])
    
    submit_spark_airflow_task_details = &quot;--conf &quot; + '\&quot;' + &quot;spark.airflowTaskDetails&quot; + &quot;=&quot; + json.dumps(airflow_task_settings) + '\'' + &quot; &quot;
    
    common_submit_args_beginning = submit_args + submit_spark_broadcast_timeout + submit_spark_max_result_size + submit_spark_dependency_jars + extra_conf + submit_spark_airflow_task_details
    
    application_resource = &quot;/update_scores.py&quot;
    application_arguments = [&quot;test_args&quot;]
    
    string_application_arguments = &quot; &quot;
    for i in range(0,len(application_arguments)):
        string_application_arguments = string_application_arguments + &quot; &quot; + json.dumps(application_arguments[i]) 
    
    
    common_submit_args_end = application_resource + string_application_arguments
            
    platform_utilities = PlatformUtilities(_infra_config)
    
    print(&quot;platform_utilities.get_python_modules_path() -&gt; &quot;,str(platform_utilities.get_python_modules_path()))
    
    submit_spark_python_module_path = &quot;--conf &quot; + '\&quot;' + &quot;spark.modulePath&quot; + &quot;=&quot; + str(platform_utilities.get_python_modules_path()) + '\&quot;' + &quot; &quot;
    
    submit_spark_args = [common_submit_args_beginning + submit_spark_pipeline_config_conf + submit_spark_python_module_path + common_submit_args_end]
    
    print(&quot;submit_spark_args -&gt; &quot;,submit_spark_args)
    
    submit_in_cluster = True
    
       
    submit_spark_pod_affinity = k8s.V1Affinity(
    node_affinity=k8s.V1NodeAffinity(k8s.V1NodeSelectorTerm(
                    match_expressions=[
                        k8s.V1NodeSelectorRequirement(key=&quot;****&quot;, operator=&quot;In&quot;, values=[&quot;n2-highmem-8&quot;]),
                        k8s.V1NodeSelectorRequirement(key=&quot;deployment&quot;, operator=&quot;In&quot;, values=[&quot;dynamic&quot;]),
                    ]
                )
            )
        )
    
    
    submit_spark_pod_tolerations = [k8s.V1Toleration(key=&quot;deployment&quot;, operator=&quot;Equal&quot;, value=&quot;dynamic&quot;, effect=&quot;NoSchedule&quot;)]
    
    application_name = &quot;test_airflow_api_test_task_id&quot;
    
    container_resources = k8s.V1ResourceRequirements(
                requests={
                    'memory': str(&quot;10Gi&quot;),
                    'cpu': str(&quot;2&quot;)
                },
                limits={
                    'memory': str(&quot;50Gi&quot;),
                    'cpu': str(&quot;5&quot;)
                }
            )
    
    submit_startup_timeout_seconds = 600
    
    submit_get_logs = True
    
    kube_submssion = KubernetesPodOperator(namespace = namespace,
                    image = docker_image,
                    is_delete_operator_pod = is_delete_operator_pod,
                    image_pull_secrets = docker_image_creds,
                    cmds = submit_command,
                    arguments = submit_spark_args,
                    in_cluster = submit_in_cluster,
                    affinity = submit_spark_pod_affinity,
                    tolerations = submit_spark_pod_tolerations,
                    container_resources = container_resources,
                    name = application_name,
                    task_id = application_name,
                    startup_timeout_seconds = submit_startup_timeout_seconds,
                    get_logs = submit_get_logs
                )
    
    kube_submssion.execute(context = None)
    
  

def def_func2(**kwargs):
    print(&quot;In func2&quot;)

dag_base = airflow_dag_object.get_dag_object()

func1=PythonOperator(
    task_id='func1',
    provide_context=True,
    python_callable=def_func1,
    dag=dag_base
)

func2=PythonOperator(
    task_id='func2',
    provide_context=True,
    python_callable=def_func2,
    dag=dag_base
)

func1 &gt;&gt; func2
</code></pre>
<p>OUTPUT ERROR:-</p>
<pre><code>Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py&quot;, line 419, in execute
    context=context,
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py&quot;, line 387, in get_or_create_pod
    pod = self.find_pod(self.namespace or pod_request_obj.metadata.namespace, context=context)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py&quot;, line 371, in find_pod
    label_selector=label_selector,
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api/core_v1_api.py&quot;, line 15697, in list_namespaced_pod
    return self.list_namespaced_pod_with_http_info(namespace, **kwargs)  # noqa: E501
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api/core_v1_api.py&quot;, line 15826, in list_namespaced_pod_with_http_info
    collection_formats=collection_formats)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api_client.py&quot;, line 353, in call_api
    _preload_content, _request_timeout, _host)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api_client.py&quot;, line 184, in __call_api
    _request_timeout=_request_timeout)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api_client.py&quot;, line 377, in request
    headers=headers)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/rest.py&quot;, line 244, in GET
    query_params=query_params)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/rest.py&quot;, line 234, in request
    raise ApiException(http_resp=r)
kubernetes.client.exceptions.ApiException: (400)
Reason: Bad Request
HTTP response headers: HTTPHeaderDict({'Audit-Id': '6ab39ea1-f955-4481-b3eb-7b3abe747a7c', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': '8e487991-120d-49d0-940a-ace0b0e64421', 'X-Kubernetes-Pf-Prioritylevel-Uid': '8f6ab0b3-abdf-4782-994c-2f0f247592d2', 'Date': 'Thu, 12 Jan 2023 13:13:20 GMT', 'Content-Length': '169'})
HTTP response body: {&quot;kind&quot;:&quot;Status&quot;,&quot;apiVersion&quot;:&quot;v1&quot;,&quot;metadata&quot;:{},&quot;status&quot;:&quot;Failure&quot;,&quot;message&quot;:&quot;found ',', expected: !, identifier, or 'end of string'&quot;,&quot;reason&quot;:&quot;BadRequest&quot;,&quot;code&quot;:400}
</code></pre>

Himanshu Malhotra
<p>In previous version of <strong>airflow &lt; 2.3</strong> <strong>KubernetesPodOperator</strong> used to work with <strong>None</strong> context</p>
<p>As mentioned in your question</p>
<pre><code>kube_submssion = KubernetesPodOperator(namespace = namespace,
                    image = docker_image,
                    is_delete_operator_pod = is_delete_operator_pod,
                    image_pull_secrets = docker_image_creds,
                    cmds = submit_command,
                    arguments = submit_spark_args,
                    in_cluster = submit_in_cluster,
                    affinity = submit_spark_pod_affinity,
                    tolerations = submit_spark_pod_tolerations,
                    container_resources = container_resources,
                    name = application_name,
                    task_id = application_name,
                    startup_timeout_seconds = submit_startup_timeout_seconds,
                    get_logs = submit_get_logs
                )
    
kube_submssion.execute(context = None)
</code></pre>
<p>The execute method is expecting the context as mentioned in the documentation at followig link</p>
<p><a href="https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes/stable/_modules/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.html#KubernetesPodOperator.execute" rel="nofollow noreferrer">https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes/stable/_modules/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.html#KubernetesPodOperator.execute</a></p>
<p>You can pass the context from **<strong>kwargs</strong> to the execute method. You can try by passing <code>kwargs</code> to <code>execute</code> method</p>
<pre><code>kube_submssion.execute(context = kwargs)
</code></pre>
