Fix KubernetesPodOperator deferrable mode with env-defined connection (#41706)

[aaron-y-chen]

Closes: #41706

Why

When users define k8s-conn-id via environment variables (e.g., AIRFLOW_CONN_...) and set deferrable=True in KubernetesPodOperator, the following exception occurs in the Triggerer:

AirflowNotFoundException: The conn_id `k8s-conn-id` isn't defined

This happens because the environment variable defining the connection exists in the Worker (where the task starts) but is missing in the Triggerer service (where the task is deferred). The Triggerer attempts to look up the connection ID but fails to find it.

How

This PR resolves the issue by resolving the connection details in the Worker before deferral and passing them explicitly to the Triggerer.

1. Operator: In KubernetesPodOperator, resolved the connection object in the execute phase (Worker) and extracted the connection extras.
2. Trigger: Added a connection_extras parameter to KubernetesPodTrigger to accept the resolved connection information.
3. Hook: Updated AsyncKubernetesHook to accept connection_extras. If provided, the hook uses these extras directly instead of attempting to look up the connection ID from the database or environment variables.

What

I used the following DAG to reproduce the issue and verify the fix in the breeze k8s environment:

importosfromdatetimeimportdatetimefromairflowimportDAGfromairflow.models.connectionimportConnectionfromairflow.providers.cncf.kubernetes.operators.podimportKubernetesPodOperatorfromairflow.providers.standard.operators.bashimportBashOperatordef_set_k8s_conn_from_env() ->None:
extra= {
"in_cluster": True,
"namespace": "airflow",
}
conn=Connection(
conn_id="k8s-conn-id",
conn_type="kubernetes",
description="K8s connection defined only via AIRFLOW_CONN_* env var",
extra=extra,
)
os.environ[f"AIRFLOW_CONN_{conn.conn_id.upper()}"] =conn.get_uri()
_set_k8s_conn_from_env()
withDAG(
dag_id="k8s_env_deferrable_repro_41706",
start_date=datetime(2024, 1, 1),
schedule=None,
catchup=False,
tags=["k8s", "deferrable", "env-conn", "issue-41706"],
) asdag:
k8s_pod=KubernetesPodOperator(
task_id="k8s_pod_env_conn_deferrable",
kubernetes_conn_id="k8s-conn-id",
namespace="airflow",
name="k8s-env-deferrable-pod",
image="alpine",
cmds=["sh", "-c"],
arguments=["mkdir -p /airflow/xcom/; echo '[1,2,3,4]' > /airflow/xcom/return.json"],
do_xcom_push=True,
get_logs=True,
deferrable=True,
on_finish_action="keep_pod",
startup_timeout_seconds=600,
)
print_xcom=BashOperator(
task_id="print_xcom_result",
bash_command="echo \"XCOM from KPO: {{ ti.xcom_pull('k8s_pod_env_conn_deferrable')[0] }}\"",
)
k8s_pod>>print_xcom

Before the fix, we can see the following error in the Airflow logs:

[2025-11-23 10:48:10] DEBUG - xcom result: [1,2,3,4] source=airflow.task.operators.airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator loc=pod.py:654 [2025-11-23 10:48:10] ERROR - Trigger emitted an error event, failing the task: The conn_id `k8s-conn-id` isn't defined source=airflow.task.operators.airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator loc=pod.py:929 [2025-11-23 10:48:11] INFO - Pod k8s-env-deferrable-pod-8k5u7y2d has phase Running source=airflow.providers.cncf.kubernetes.utils.pod_manager.PodManager loc=pod_manager.py:763

After the fix, we can see the new log we added in the Triggerer logs, and this error log no longer appears.

[Nataneljpwd]

Minor nit picks, looks very good.

[aaron-y-chen]

All the conflicts are resolved. If there is anything that can be improved, please don't hesitate to let me know :D