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 5f03059ee9b fix: coerce explicit None _request_timeout in async/sync
k8s API clients (#69611)
5f03059ee9b is described below
commit 5f03059ee9b888c1d31268e66c0ad1c8bb8b9554
Author: Mingjie Zhao <[email protected]>
AuthorDate: Thu Aug 13 20:34:56 2026 +0800
fix: coerce explicit None _request_timeout in async/sync k8s API clients
(#69611)
The _TimeoutAsyncK8sApiClient and _TimeoutK8sApiClient wrappers attempt
to enforce a client-side read timeout via kwargs.setdefault(). However,
every generated kubernetes / kubernetes_asyncio API method (e.g.
read_namespaced_pod) always forwards _request_timeout=None explicitly:
_request_timeout=local_var_params.get('_request_timeout') # → None
setdefault() is a no-op when the key already exists, even as None. So
None reaches aiohttp, which creates ClientTimeout() with no total or
sock_read limit — an infinite socket timeout.
On a half-open TCP connection (what a 429 Too Many Requests or an
LB-side "500 context canceled" leaves behind when the LB drops the
socket without FIN/RST), the awaited API call blocks forever. Because
no exception is raised, generic_api_retry never fires and a deferrable
KubernetesPodOperator trigger hangs indefinitely with no error visible
in the task log. Only a triggerer restart clears it.
Fix: replace setdefault with an explicit None-check so that a None
value (from the generated client) is correctly treated as "not set",
while any intentionally-set value — including the (1800, 300) tuple
passed by the async log-streaming path — is preserved.
Confirmed by live introspection in production (kubernetes_asyncio 35.0.1,
aiohttp 3.13.2, apache-airflow-providers-cncf-kubernetes 10.17.1):
before the fix, _request_timeout=None was forwarded to aiohttp regardless
of the wrapper; after the fix, it correctly receives 60.
Ref:
https://github.com/kubernetes-client/python/blob/master/examples/watch/timeout-settings.md
Co-authored-by: Mingjie Zhao <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../airflow/providers/cncf/kubernetes/kube_client.py | 19 +++++++++++++------
.../unit/cncf/kubernetes/hooks/test_kubernetes.py | 10 ++++++++++
2 files changed, 23 insertions(+), 6 deletions(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
index 79e2b802e01..08f91abe73e 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
@@ -79,9 +79,12 @@ try:
def call_api(self, *args: Any, **kwargs: Any) -> Any:
timeout_seconds = kwargs.get("timeout_seconds") # get server-side
timeout
- kwargs.setdefault(
- "_request_timeout", _get_request_timeout(timeout_seconds)
- ) # client-side timeout
+ # Use setdefault's intent but handle explicit None: generated
kubernetes client methods always
+ # pass _request_timeout=None explicitly, making setdefault a no-op
and leaving an infinite
+ # timeout. On a half-open TCP connection (e.g. after a 429 or
LB-side 500), this causes the
+ # call to hang indefinitely with no exception, preventing any
retry or recovery.
+ if kwargs.get("_request_timeout") is None:
+ kwargs["_request_timeout"] =
_get_request_timeout(timeout_seconds)
return super().call_api(*args, **kwargs)
class _TimeoutAsyncK8sApiClient(async_client.ApiClient):
@@ -95,9 +98,13 @@ try:
async def call_api(self, *args: Any, **kwargs: Any) -> Any:
timeout_seconds = kwargs.get("timeout_seconds") # server-side
timeout
- kwargs.setdefault(
- "_request_timeout", _get_request_timeout(timeout_seconds)
- ) # client-side timeout
+ # Use setdefault's intent but handle explicit None: generated
kubernetes_asyncio client methods
+ # always pass _request_timeout=None explicitly, making setdefault
a no-op and leaving an
+ # infinite timeout. On a half-open TCP connection (e.g. after a
429 or LB-side 500), this
+ # causes the trigger's await to hang indefinitely with no
exception, preventing any retry or
+ # recovery.
+ if kwargs.get("_request_timeout") is None:
+ kwargs["_request_timeout"] =
_get_request_timeout(timeout_seconds)
return await super().call_api(*args, **kwargs)
except ImportError as e:
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 7105cfad845..09d686821ca 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
@@ -94,7 +94,12 @@ class TestTimeoutK8sApiClient:
("kwargs", "expected_timeout"),
[
pytest.param({}, API_TIMEOUT, id="default-timeout"),
+ # Generated kubernetes client methods always pass
_request_timeout=None explicitly,
+ # so setdefault() is a no-op. The fix must coerce None to the
default timeout.
+ pytest.param({"_request_timeout": None}, API_TIMEOUT,
id="explicit-none-timeout"),
pytest.param({"timeout_seconds": 5678, "_request_timeout": 1234},
1234, id="explicit-timeout"),
+ # Log-streaming path passes a (connection, read) tuple; it must
not be clobbered.
+ pytest.param({"_request_timeout": (1800, 300)}, (1800, 300),
id="explicit-tuple-preserved"),
pytest.param(
{"timeout_seconds": API_TIMEOUT -
API_TIMEOUT_OFFSET_SERVER_SIDE},
API_TIMEOUT,
@@ -128,7 +133,12 @@ class TestTimeoutAsyncK8sApiClient:
("kwargs", "expected_timeout"),
[
pytest.param({}, API_TIMEOUT, id="default-timeout"),
+ # Generated kubernetes_asyncio client methods always pass
_request_timeout=None explicitly,
+ # so setdefault() is a no-op. The fix must coerce None to the
default timeout.
+ pytest.param({"_request_timeout": None}, API_TIMEOUT,
id="explicit-none-timeout"),
pytest.param({"timeout_seconds": 5678, "_request_timeout": 1234},
1234, id="explicit-timeout"),
+ # Log-streaming path passes a (connection, read) tuple; it must
not be clobbered.
+ pytest.param({"_request_timeout": (1800, 300)}, (1800, 300),
id="explicit-tuple-preserved"),
pytest.param(
{"timeout_seconds": API_TIMEOUT -
API_TIMEOUT_OFFSET_SERVER_SIDE},
API_TIMEOUT,
