Reuse async Kubernetes API client across trigger polls

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py

@@ -840,6 +840,7 @@ def __init__(
         # Cached result of exec-auth detection. None means not yet detected.
         # This is to optimise and not calling _uses_exec_auth repeatedly on every _load_config() call.
         self._is_exec_auth: bool | None = None
+        self._cached_kube_client: async_client.ApiClient | None = None
 
     def _uses_exec_auth(self, kubeconfig_data: dict, context: str | None = None) -> bool:
         """
@@ -1024,14 +1025,28 @@ async def _get_field(self, field_name):
 
     @contextlib.asynccontextmanager
     async def get_conn(self) -> AsyncGenerator[async_client.ApiClient, None]:
-        kube_client = None
+        await self._load_config()
+        if self._config_loaded:
+            # Reuse one client per hook: each construction runs ssl.create_default_context()
+            # on the event loop and opens a new connection pool. Owners release it via
+            # close(); triggers do so in cleanup().
+            if self._cached_kube_client is None:
+                self._cached_kube_client = _TimeoutAsyncK8sApiClient(configuration=self.client_configuration)
+            yield self._cached_kube_client
+            return
+        # Exec-based auth rotates short-lived tokens by reloading the config on
+        # every call, so the client cannot be reused.
+        kube_client = _TimeoutAsyncK8sApiClient(configuration=self.client_configuration)
         try:
-            await self._load_config()
-            kube_client = _TimeoutAsyncK8sApiClient(configuration=self.client_configuration)
             yield kube_client
         finally:
-            if kube_client is not None:
-                await kube_client.close()
+            await kube_client.close()
+
+    async def close(self) -> None:
+        """Release the cached API client, if any. Safe to call multiple times."""
+        cached, self._cached_kube_client = self._cached_kube_client, None
+        if cached is not None:
+            await cached.close()
 
     @generic_api_retry
     async def get_pod(self, name: str, namespace: str) -> V1Pod:

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/job.py

@@ -174,3 +174,8 @@ def pod_manager(self) -> PodManager:
             cluster_context=self.cluster_context,
         )
         return PodManager(kube_client=sync_hook.core_v1_client)
+
+    async def cleanup(self) -> None:
+        """Release the hook's cached API client when the trigger exits."""
+        await super().cleanup()
+        await self.hook.close()

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py

@@ -538,46 +538,54 @@ async def cleanup(self) -> None:
         On Airflow 3.3.0+ pod deletion on user kill is handled in ``on_kill()`` only; this avoids
         deleting pods on triggerer restart. On older Airflow versions, ``cleanup()`` still uses
         ``safe_to_cancel()`` because ``on_kill()`` is not wired for user kills.
-        """
-        # TODO: Remove this Airflow < 3.3 cleanup branch (early return, ``safe_to_cancel``, and
-        # related tests) once the minimum Airflow version supported by this provider is >= 3.3.
-        # In Airflow 3.3+, ``BaseTrigger.on_kill()`` handles user-initiated kills; keeping the
-        # legacy path for backward compatibility with older Airflow versions.
-        if AIRFLOW_V_3_3_PLUS:
-            return
-
-        if self._fired_event:
-            self.log.debug("Skipping cleanup since an event has already been fired.")
-            return
-
-        if self.on_kill_action == OnKillAction.KEEP_POD:
-            self.log.debug("Skipping cleanup since on_kill_action is set to %r.", self.on_kill_action.value)
-            return
 
+        Always releases the hook's cached API client, since the triggerer invokes ``cleanup()``
+        after the trigger finishes for any reason (event fired, cancelled, or killed).
+        """
         try:
-            safe = await self.safe_to_cancel()
-        except Exception:
-            self.log.warning(
-                "Could not determine task state during cleanup; skipping pod deletion to be safe.",
-                exc_info=True,
-            )
-            return
+            # TODO: Remove this Airflow < 3.3 cleanup branch (early return, ``safe_to_cancel``, and
+            # related tests) once the minimum Airflow version supported by this provider is >= 3.3.
+            # In Airflow 3.3+, ``BaseTrigger.on_kill()`` handles user-initiated kills; keeping the
+            # legacy path for backward compatibility with older Airflow versions.
+            if AIRFLOW_V_3_3_PLUS:
+                return
+
+            if self._fired_event:
+                self.log.debug("Skipping cleanup since an event has already been fired.")
+                return
+
+            if self.on_kill_action == OnKillAction.KEEP_POD:
+                self.log.debug(
+                    "Skipping cleanup since on_kill_action is set to %r.", self.on_kill_action.value
+                )
+                return
 
-        if not safe:
-            self.log.debug(
-                "Skipping cleanup since the task is still in deferred state (likely a triggerer restart)."
-            )
-            return
+            try:
+                safe = await self.safe_to_cancel()
+            except Exception:
+                self.log.warning(
+                    "Could not determine task state during cleanup; skipping pod deletion to be safe.",
+                    exc_info=True,
+                )
+                return
 
-        self.log.info("Deleting pod %s in namespace %s.", self.pod_name, self.pod_namespace)
-        try:
-            await self.hook.delete_pod(
-                name=self.pod_name,
-                namespace=self.pod_namespace,
-                grace_period_seconds=self.termination_grace_period,
-            )
-        except Exception:
-            self.log.exception("Unexpected error while deleting pod %s", self.pod_name)
+            if not safe:
+                self.log.debug(
+                    "Skipping cleanup since the task is still in deferred state (likely a triggerer restart)."
+                )
+                return
+
+            self.log.info("Deleting pod %s in namespace %s.", self.pod_name, self.pod_namespace)
+            try:
+                await self.hook.delete_pod(
+                    name=self.pod_name,
+                    namespace=self.pod_namespace,
+                    grace_period_seconds=self.termination_grace_period,
+                )
+            except Exception:
+                self.log.exception("Unexpected error while deleting pod %s", self.pod_name)
+        finally:
+            await self.hook.close()
 
     def define_container_state(self, pod: V1Pod) -> ContainerState:
         if pod.status is None or pod.status.container_statuses is None: