fix templating of KubernetesPodOperator env_vars field

[nyoungstudios]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.6.3

What happened?

I happen to be using an older version of Airflow, but I think it still relevant for newer versions.

Basically, this issue: #25827 and this pr: #25829

The type check happens in the __init__ function before the jinja template strings (or XComArg) gets rendered.

What you think should happen instead?

But, I don't think the pr is how I would have solved it. There is no need to call ast.literal_eval() as pointed out in the pr review.

More simply moving this check

airflow/airflow/providers/cncf/kubernetes/operators/pod.py

Lines 333 to 336 in 5df7a1e

	env_vars = convert_env_vars(env_vars) if env_vars else []
	self.env_vars = env_vars
	if pod_runtime_info_envs:
	self.env_vars.extend([convert_pod_runtime_info_env(p) for p in pod_runtime_info_envs])

from the __init__ function (when the jinja templating strings (or XComArg) do not get rendered) to a function that has access to the Python context variable (after the jinja templating string (or XComArg) gets rendered). This could be the pre_execute function, but I don't think that is a good idea in case someone wants to inherit from this operator and implement the pre_execute function - they would have to remember to call super().pre_execute(context). So, I think it could be more simply implemented in

airflow/airflow/providers/cncf/kubernetes/operators/pod.py

Line 979 in 5df7a1e

def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:

function which is called in the execute function.

How to reproduce

Here is a sample dag using a dynamic task group that works on the BashOperator but not on the KubernetesPodOperator:

from airflow.decorators import task_group
from airflow.models import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

with DAG(
    dag_id="test",
    description="Tests dynamic task mapping bash vs k8s operator",
    schedule_interval=None,
    catchup=False,
) as dag:

    items = PythonOperator(
        task_id="get-items",
        python_callable=lambda: ["apple", "banana", "cherry"],
    )

    @task_group
    def steps_transforms(value: str):
        upper = PythonOperator(
            task_id="upper",
            python_callable=lambda x: {"VALUE": x.upper()},
            op_args=[value],
        )

        # this works
        echo = BashOperator(
            task_id="echo",
            bash_command="echo $VALUE",
            env=upper.output,
        )

        # this doesn't work
        echo_on_k8s = KubernetesPodOperator(
            task_id="echo-on-k8s",
            image="docker-image-uri",
            cmds=["/bin/bash"],
            arguments=[
                "-ce",
                "echo $VALUE",
            ],
            name="echo-on-k8s",
            is_delete_operator_pod=True,
            in_cluster=False,
            env_vars=upper.output,
        )

        upper >> echo
        upper >> echo_on_k8s


    items >> steps_transforms.expand(value=items.output)

Operating System

Debian 11

Versions of Apache Airflow Providers

apache-airflow-providers-apache-beam==5.3.0
apache-airflow-providers-cncf-kubernetes==7.3.0
apache-airflow-providers-common-sql==1.8.0
apache-airflow-providers-dbt-cloud==3.4.0
apache-airflow-providers-ftp==3.6.0
apache-airflow-providers-google==10.10.0
apache-airflow-providers-hashicorp==3.5.0
apache-airflow-providers-http==4.6.0
apache-airflow-providers-imap==3.4.0
apache-airflow-providers-mysql==5.2.0
apache-airflow-providers-postgres==5.7.0
apache-airflow-providers-sendgrid==3.3.0
apache-airflow-providers-sftp==4.7.0
apache-airflow-providers-slack==8.3.0
apache-airflow-providers-sqlite==3.5.0
apache-airflow-providers-ssh==3.8.0

Deployment

Google Cloud Composer

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[nyoungstudios]

@potiuk I ran into this same problem of a pr that you previously reviewed and closed due to a bad implementation. I was just trying to use the KubernetesPodOperator (or more specifically the GKEStartPodOperator since I was running it on Google Cloud Composer) with dynamic task mapping within a task group.

Anyway, I haven't officially tried out my proposed solution as I need to setup my dev environment to contribute to this repo, but I think I can solve this problem of the previous closed issue/pr.

[potiuk]

Just try it in PR and add tests. There is no reason/need to open issue for it - it makes an extra cognitive load on maintainers to try to understand what you want to do rather than you showing what you want to do in the PR (and tests covering it).

Converting it to a discussion as this is not an issue

[nyoungstudios]

Thanks, I'll do that next time. Sorry for the extra noise.

Issue fixed in this pr: #39139