You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "nailo2c (via GitHub)" <gi...@apache.org> on 2025/11/29 02:47:11 UTC
[PR] Fix `KubernetesPodOperator` deferrable mode with env-defined connection (#41706) [airflow]
nailo2c opened a new pull request, #58841:
URL: https://github.com/apache/airflow/pull/58841
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:
```console
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 `conn_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:
```python
import os
from datetime import datetime
from airflow import DAG
from airflow.models.connection import Connection
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from airflow.providers.standard.operators.bash import BashOperator
def _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()
with DAG(
dag_id="k8s_env_deferrable_repro_41706",
start_date=datetime(2024, 1, 1),
schedule=None,
catchup=False,
tags=["k8s", "deferrable", "env-conn", "issue-41706"],
) as dag:
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:
```console
[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.
<img width="1910" height="989" alt="defer_example" src="https://github.com/user-attachments/assets/b941b67a-da8b-446b-aff2-91ede4d64f02" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix `KubernetesPodOperator` deferrable mode with env-defined connection (#41706) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #58841:
URL: https://github.com/apache/airflow/pull/58841#discussion_r2575770929
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -856,13 +857,28 @@ def convert_config_file_to_dict(self):
def invoke_defer_method(self, last_log_time: DateTime | None = None) -> None:
"""Redefine triggers which are being used in child classes."""
self.convert_config_file_to_dict()
+
+ conn_extras = None
+ if self.kubernetes_conn_id:
+ try:
+ conn = BaseHook.get_connection(self.kubernetes_conn_id)
+ conn_extras = conn.extra_dejson
+ self.log.info("Successfully resolved connection extras for deferral.")
+ except Exception as e:
+ self.log.warning(
+ "Could not resolve connection extras for deferral: %s. "
+ "Triggerer will try to resolve it from its own environment.",
+ e,
+ )
+
trigger_start_time = datetime.datetime.now(tz=datetime.timezone.utc)
self.defer(
trigger=KubernetesPodTrigger(
pod_name=self.pod.metadata.name, # type: ignore[union-attr]
pod_namespace=self.pod.metadata.namespace, # type: ignore[union-attr]
trigger_start_time=trigger_start_time,
kubernetes_conn_id=self.kubernetes_conn_id,
+ connection_extras=conn_extras,
Review Comment:
Hi, thanks for the review. I'll fix it and keep the naming consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix `KubernetesPodOperator` deferrable mode with env-defined connection (#41706) [airflow]
Posted by "romsharon98 (via GitHub)" <gi...@apache.org>.
romsharon98 merged PR #58841:
URL: https://github.com/apache/airflow/pull/58841
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix `KubernetesPodOperator` deferrable mode with env-defined connection (#41706) [airflow]
Posted by "Nataneljpwd (via GitHub)" <gi...@apache.org>.
Nataneljpwd commented on code in PR #58841:
URL: https://github.com/apache/airflow/pull/58841#discussion_r2574910159
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -856,13 +857,28 @@ def convert_config_file_to_dict(self):
def invoke_defer_method(self, last_log_time: DateTime | None = None) -> None:
"""Redefine triggers which are being used in child classes."""
self.convert_config_file_to_dict()
+
+ conn_extras = None
+ if self.kubernetes_conn_id:
+ try:
+ conn = BaseHook.get_connection(self.kubernetes_conn_id)
+ conn_extras = conn.extra_dejson
+ self.log.info("Successfully resolved connection extras for deferral.")
+ except Exception as e:
Review Comment:
Shouldn't we specify the exception? What if the exception is a connection refused or timeout? We do not want to pass that to the triggerer
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -856,13 +857,28 @@ def convert_config_file_to_dict(self):
def invoke_defer_method(self, last_log_time: DateTime | None = None) -> None:
"""Redefine triggers which are being used in child classes."""
self.convert_config_file_to_dict()
+
+ conn_extras = None
+ if self.kubernetes_conn_id:
+ try:
+ conn = BaseHook.get_connection(self.kubernetes_conn_id)
+ conn_extras = conn.extra_dejson
+ self.log.info("Successfully resolved connection extras for deferral.")
+ except Exception as e:
+ self.log.warning(
+ "Could not resolve connection extras for deferral: %s. "
+ "Triggerer will try to resolve it from its own environment.",
+ e,
+ )
+
trigger_start_time = datetime.datetime.now(tz=datetime.timezone.utc)
self.defer(
trigger=KubernetesPodTrigger(
pod_name=self.pod.metadata.name, # type: ignore[union-attr]
pod_namespace=self.pod.metadata.namespace, # type: ignore[union-attr]
trigger_start_time=trigger_start_time,
kubernetes_conn_id=self.kubernetes_conn_id,
+ connection_extras=conn_extras,
Review Comment:
Why not keep it as either `conn_extras` or `connection_extras` all the way through?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix `KubernetesPodOperator` deferrable mode with env-defined connection (#41706) [airflow]
Posted by "prdai (via GitHub)" <gi...@apache.org>.
prdai commented on code in PR #58841:
URL: https://github.com/apache/airflow/pull/58841#discussion_r2575726689
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -856,13 +857,28 @@ def convert_config_file_to_dict(self):
def invoke_defer_method(self, last_log_time: DateTime | None = None) -> None:
"""Redefine triggers which are being used in child classes."""
self.convert_config_file_to_dict()
+
+ conn_extras = None
+ if self.kubernetes_conn_id:
+ try:
+ conn = BaseHook.get_connection(self.kubernetes_conn_id)
+ conn_extras = conn.extra_dejson
+ self.log.info("Successfully resolved connection extras for deferral.")
+ except Exception as e:
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix `KubernetesPodOperator` deferrable mode with env-defined connection (#41706) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on PR #58841:
URL: https://github.com/apache/airflow/pull/58841#issuecomment-3644952314
All the conflicts are resolved. If there is anything that can be improved, please don't hesitate to let me know :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix `KubernetesPodOperator` deferrable mode with env-defined connection (#41706) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #58841:
URL: https://github.com/apache/airflow/pull/58841#discussion_r2575871093
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -856,13 +857,28 @@ def convert_config_file_to_dict(self):
def invoke_defer_method(self, last_log_time: DateTime | None = None) -> None:
"""Redefine triggers which are being used in child classes."""
self.convert_config_file_to_dict()
+
+ conn_extras = None
+ if self.kubernetes_conn_id:
+ try:
+ conn = BaseHook.get_connection(self.kubernetes_conn_id)
+ conn_extras = conn.extra_dejson
+ self.log.info("Successfully resolved connection extras for deferral.")
+ except Exception as e:
Review Comment:
Good point, that's a great way to avoid confusion. I'll fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org