From fd52db49a20d22ebcd69c2dd7eaaa2de2319738c Mon Sep 17 00:00:00 2001 From: Mingjie Zhao Date: Wed, 8 Jul 2026 20:45:06 +0800 Subject: [PATCH] fix: coerce explicit None _request_timeout in async/sync k8s API clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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: Claude Sonnet 4.6 --- .../providers/cncf/kubernetes/hooks/kubernetes.py | 12 ++++++++++-- .../unit/cncf/kubernetes/hooks/test_kubernetes.py | 10 ++++++++++ 2 files changed, 20 insertions(+), 2 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 9dcb6b59bba0a..5715a605a31c0 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 @@ -104,7 +104,10 @@ def __init__( def call_api(self, *args, **kwargs): 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. + if kwargs.get("_request_timeout") is None: + kwargs["_request_timeout"] = _get_request_timeout(timeout_seconds) return super().call_api(*args, **kwargs) @@ -119,7 +122,12 @@ def __init__( 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 + # 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) 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 de9670f490371..e96c35b93d2a7 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 @@ -93,7 +93,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, @@ -127,7 +132,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,