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,