ZhaoMJ opened a new pull request, #69611:
URL: https://github.com/apache/airflow/pull/69611
## 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:
```python
_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.
```python
# 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](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
guide
- [x] My PR is limited to one change (bug fix)
- [x] I have added tests that prove my fix is effective
- [x] All existing tests pass (verified locally for the affected test
classes)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
