This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e2cf316fc36 Fix race condition in AsyncKubernetesHook corrupting
global kubernetes_asyncio config (#65566)
e2cf316fc36 is described below
commit e2cf316fc36c674e5af7bce768b8a2e5a2bb1b6f
Author: Mingjie Zhao <[email protected]>
AuthorDate: Sun May 10 04:27:26 2026 +0200
Fix race condition in AsyncKubernetesHook corrupting global
kubernetes_asyncio config (#65566)
* Fix race condition in AsyncKubernetesHook corrupting global
kubernetes_asyncio config
When multiple deferrable KubernetesPodOperator tasks run concurrently on the
same triggerer, each task's AsyncKubernetesHook calls
load_kube_config_from_dict
(or load_incluster_config), which writes to the module-level global
kubernetes_asyncio Configuration. Because load_kube_config_from_dict is
async
and contains multiple await points, two coroutines for different clusters
can
interleave: one writes its server URL while the other writes its bearer
token,
producing a mixed configuration. _TimeoutAsyncK8sApiClient() was then
instantiated with no configuration= argument, so it silently consumed
whatever
global state happened to be current at that moment.
The fix has four parts:
- Add __init__ to _TimeoutAsyncK8sApiClient accepting configuration=
(matching
the existing sync _TimeoutK8sApiClient pattern).
- Initialise self.client_configuration to a fresh
async_client.Configuration()
at the top of _load_config when it is None, ensuring each hook instance
owns
an isolated Configuration object.
- Pass client_configuration=self.client_configuration to
load_incluster_config
and load_kube_config_from_dict, which previously wrote only to global
state.
- Pass configuration=self.client_configuration when constructing
_TimeoutAsyncK8sApiClient in get_conn, so the client uses the per-instance
configuration rather than the global default.
The kubeconfig_path, kube_config (tempfile), and default-file branches
already
forwarded client_configuration=self.client_configuration; they now benefit
from
the initialisation fix so the value is never None.
* Add tests for AsyncKubernetesHook per-instance Configuration isolation
Covers the race condition fix across five scenarios:
- _TimeoutAsyncK8sApiClient accepts a configuration= argument and forwards
it
to the parent ApiClient, ensuring each instance is isolated.
- No-arg construction still works for backwards compatibility.
- _load_config allocates a fresh async_client.Configuration() per hook
instance
when none was supplied by the caller.
- load_incluster_config and load_kube_config_from_dict are each called with
the
per-instance client_configuration rather than leaving it as None (which
would
cause both functions to fall back to Configuration.set_default() and
corrupt
shared global state).
- get_conn passes configuration= to _TimeoutAsyncK8sApiClient so the API
client
uses the hook's isolated Configuration, not the global default.
- The core regression test runs two hooks for different clusters
concurrently
via asyncio.gather and asserts that each hook's client_configuration is a
distinct object populated only with that hook's own connection data.
* Address review: override client_configuration type in AsyncKubernetesHook
The parent KubernetesHook.client_configuration is typed as
kubernetes.client.Configuration (sync). Rather than using a separate
attribute, AsyncKubernetesHook now explicitly accepts and re-declares
client_configuration as async_client.Configuration | None, making the
override visible to type checkers and callers without adding a new
attribute to track.
Both Configuration classes are generated from the same OpenAPI source and
are structurally compatible when library versions align.
Also add an await asyncio.sleep(0) yield inside the populate_config mock
side effect in the concurrency regression test so asyncio.gather can
actually interleave the two coroutines and exercise the race window.
---------
Co-authored-by: Mingjie Zhao <[email protected]>
---
.../providers/cncf/kubernetes/hooks/kubernetes.py | 23 ++++-
.../unit/cncf/kubernetes/hooks/test_kubernetes.py | 112 +++++++++++++++++++++
2 files changed, 132 insertions(+), 3 deletions(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index 3f740981aa7..9dcb6b59bba 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -111,6 +111,12 @@ class _TimeoutK8sApiClient(client.ApiClient):
class _TimeoutAsyncK8sApiClient(async_client.ApiClient):
"""Wrapper around kubernetes async ApiClient to set default timeout."""
+ def __init__(
+ self,
+ configuration: async_client.Configuration | None = None,
+ ) -> None:
+ super().__init__(configuration=configuration)
+
async def call_api(self, *args, **kwargs):
timeout_seconds = kwargs.get("timeout_seconds") # server-side timeout
kwargs.setdefault("_request_timeout",
_get_request_timeout(timeout_seconds)) # client-side timeout
@@ -847,7 +853,12 @@ class AsyncKubernetesHook(KubernetesHook):
"""Hook to use Kubernetes SDK asynchronously."""
def __init__(
- self, config_dict: dict | None = None, connection_extras: dict | None
= None, *args, **kwargs
+ self,
+ config_dict: dict | None = None,
+ connection_extras: dict | None = None,
+ client_configuration: async_client.Configuration | None = None,
+ *args,
+ **kwargs,
):
super().__init__(*args, **kwargs)
@@ -855,6 +866,8 @@ class AsyncKubernetesHook(KubernetesHook):
self._extras: dict | None = connection_extras
self._event_polling_fallback = False
self._config_loaded = False
+ # Override the parent's sync-typed client_configuration with the async
type.
+ self.client_configuration: async_client.Configuration | None =
client_configuration
# 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
@@ -894,6 +907,9 @@ class AsyncKubernetesHook(KubernetesHook):
and cached. For exec-based auth (EKS, GKE), the config is reloaded on
every call so
that short-lived tokens are always refreshed.
"""
+ if self.client_configuration is None:
+ self.client_configuration = async_client.Configuration()
+
if self._config_loaded:
return
@@ -915,7 +931,7 @@ class AsyncKubernetesHook(KubernetesHook):
if in_cluster:
self.log.debug(LOADING_KUBE_CONFIG_FILE_RESOURCE.format("within a
pod"))
- async_config.load_incluster_config()
+
async_config.load_incluster_config(client_configuration=self.client_configuration)
self._is_in_cluster = True
self._config_loaded = True
return
@@ -927,6 +943,7 @@ class AsyncKubernetesHook(KubernetesHook):
await async_config.load_kube_config_from_dict(
self.config_dict,
context=cluster_context,
+ client_configuration=self.client_configuration,
)
if self._is_exec_auth is None:
@@ -1041,7 +1058,7 @@ class AsyncKubernetesHook(KubernetesHook):
kube_client = None
try:
await self._load_config()
- kube_client = _TimeoutAsyncK8sApiClient()
+ kube_client =
_TimeoutAsyncK8sApiClient(configuration=self.client_configuration)
yield kube_client
finally:
if kube_client is not None:
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
index 2d38c697954..de9670f4903 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import asyncio
import json
import os
import tempfile
@@ -155,6 +156,22 @@ class TestTimeoutAsyncK8sApiClient:
assert call_kwargs["_request_timeout"] == expected_timeout
assert out == "ok"
+ @pytest.mark.asyncio
+ async def test_accepts_configuration_parameter(self):
+ """Constructor forwards configuration= to ApiClient so each instance
is isolated."""
+ config = async_client.Configuration()
+ config.host = "https://test-cluster.example.com";
+ client = _TimeoutAsyncK8sApiClient(configuration=config)
+ assert client.configuration is config
+ await client.close()
+
+ @pytest.mark.asyncio
+ async def test_no_arg_construction_works(self):
+ """No-arg construction must still work for backwards compatibility."""
+ client = _TimeoutAsyncK8sApiClient()
+ assert client is not None
+ await client.close()
+
@pytest.fixture
def remove_default_conn(monkeypatch):
@@ -1935,3 +1952,98 @@ class TestAsyncKubernetesHook:
]
)
mock_sleep.assert_awaited_once_with(10)
+
+ # -------------------------------------------------------------------------
+ # Tests for the per-instance Configuration isolation fix
+ # -------------------------------------------------------------------------
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_load_config_initialises_client_configuration(self,
mock_load):
+ """_load_config allocates a fresh per-instance Configuration when none
was supplied."""
+ mock_load.return_value = None
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict={"a": "b"})
+ assert hook.client_configuration is None
+
+ await hook._load_config()
+
+ assert isinstance(hook.client_configuration,
async_client.Configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_incluster_config")
+ async def test_load_config_incluster_passes_client_configuration(self,
mock_incluster):
+ """load_incluster_config is called with the per-instance
client_configuration."""
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=True)
+
+ await hook._load_config()
+
+
mock_incluster.assert_called_once_with(client_configuration=hook.client_configuration)
+ assert isinstance(hook.client_configuration,
async_client.Configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_load_config_config_dict_passes_client_configuration(self,
mock_load):
+ """load_kube_config_from_dict is called with the per-instance
client_configuration."""
+ mock_load.return_value = None
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict={"a": "b"})
+
+ await hook._load_config()
+
+ mock_load.assert_called_once()
+ _, kwargs = mock_load.call_args
+ assert "client_configuration" in kwargs
+ assert kwargs["client_configuration"] is hook.client_configuration
+ assert isinstance(hook.client_configuration,
async_client.Configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_get_conn_passes_client_configuration_to_async_client(self,
mock_load):
+ """get_conn passes the per-instance configuration= to
_TimeoutAsyncK8sApiClient."""
+ mock_load.return_value = None
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict={"a": "b"})
+
+ with mock.patch(HOOK_MODULE + "._TimeoutAsyncK8sApiClient") as
mock_client_cls:
+ mock_client_cls.return_value.close = mock.AsyncMock()
+ async with hook.get_conn():
+ pass
+
+
mock_client_cls.assert_called_once_with(configuration=hook.client_configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_concurrent_hooks_have_isolated_configurations(self,
mock_load):
+ """Two concurrent hooks for different clusters each use an independent
Configuration.
+
+ This is the core regression test for the race condition: before the
fix,
+ load_kube_config_from_dict was called with client_configuration=None,
causing it
+ to overwrite the global kubernetes_asyncio Configuration via
Configuration.set_default().
+ Concurrent triggerer coroutines for different clusters would
interleave at the await
+ inside load_kube_config_from_dict and produce a mixed global state
(e.g. cluster-a's
+ server URL paired with cluster-b's bearer token).
+ """
+
+ async def populate_config(config_dict, context, client_configuration):
+ # Yield to the event loop so concurrent coroutines can interleave,
+ # simulating the await points inside the real
load_kube_config_from_dict.
+ await asyncio.sleep(0)
+ client_configuration.host = config_dict["server"]
+
+ mock_load.side_effect = populate_config
+
+ hook_a = AsyncKubernetesHook(
+ conn_id=None,
+ in_cluster=False,
+ config_dict={"server": "https://cluster-a.example.com"},
+ )
+ hook_b = AsyncKubernetesHook(
+ conn_id=None,
+ in_cluster=False,
+ config_dict={"server": "https://cluster-b.example.com"},
+ )
+
+ await asyncio.gather(hook_a._load_config(), hook_b._load_config())
+
+ # Each hook must own a separate, non-contaminated Configuration object.
+ assert hook_a.client_configuration is not hook_b.client_configuration
+ assert hook_a.client_configuration.host ==
"https://cluster-a.example.com";
+ assert hook_b.client_configuration.host ==
"https://cluster-b.example.com";
