Coerce explicit None _request_timeout in async/sync k8s API clients

[ZhaoMJ]

Summary

The _TimeoutAsyncK8sApiClient and _TimeoutK8sApiClient wrappers were introduced to enforce a 60s client-side read timeout on all Kubernetes API calls. However, the timeout is silently dropped, leaving an infinite socket read timeout that causes deferrable KubernetesPodOperator triggers to hang indefinitely.

Root cause

Every generated kubernetes / kubernetes_asyncio API method (e.g. read_namespaced_pod) always forwards _request_timeout explicitly as None when the caller doesn't set it:

_request_timeout=local_var_params.get('_request_timeout')  # → None

The wrapper tried to inject the default using kwargs.setdefault("_request_timeout", ...). But 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 load balancer 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 KPO trigger hangs indefinitely with no error visible in the task log. Only a triggerer restart clears it.

This is confirmed by the official kubernetes Python client documentation:

"During network outages (dropped packets, no RST/FIN), [server-side timeout] has no effect — the client will hang indefinitely without an exception."
"If [client-side _request_timeout] not set, defaults to None — meaning no socket timeout."

Ref: https://github.com/kubernetes-client/python/blob/master/examples/watch/timeout-settings.md

Fix

Replace setdefault with an explicit None-check, so a None value (from the generated client) is treated as "not set" and receives the 60s default, while any intentionally-set value is preserved — including the (1800, 300) tuple passed by the async log-streaming path, which must not be clobbered.

# Before:
kwargs.setdefault("_request_timeout", _get_request_timeout(timeout_seconds))

# After:
if kwargs.get("_request_timeout") is None:
    kwargs["_request_timeout"] = _get_request_timeout(timeout_seconds)

Verification

Confirmed by live introspection in a production triggerer pod (kubernetes_asyncio==35.0.1, aiohttp==3.13.2, apache-airflow-providers-cncf-kubernetes==10.17.1):

DEFAULT_TIMEOUT_WOULD_BE: 60
ACTUALLY_FORWARDED_TO_AIOHTTP: None   ← before fix
ACTUALLY_FORWARDED_TO_AIOHTTP: 60    ← after fix

All cases validated:

• Explicit None (generated client path) → 60s ✓
• Key absent → 60s ✓
• Explicit (1800, 300) tuple (log streaming) → preserved ✓
• timeout_seconds scaling → intact ✓

Tests

Added two new parametrize cases to both TestTimeoutK8sApiClient and TestTimeoutAsyncK8sApiClient:

• explicit-none-timeout: the actual failure mode (generated client always passes None)
• explicit-tuple-preserved: the log-streaming invariant ((1800, 300) must not be overwritten)

The existing default-timeout case ({}) only covers the key-absent path, not the key-present-as-None path that the generated client always exercises.

Checklist

• I have read the CONTRIBUTING.md guide
• My PR is limited to one change (bug fix)
• I have added tests that prove my fix is effective
• All existing tests pass (verified locally for the affected test classes)

🤖 Generated with Claude Code