You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "AutomationDev85 (via GitHub)" <gi...@apache.org> on 2026/01/08 09:21:47 UTC
[PR] KubernetesPodOperator: Fix hanging API calls [airflow]
AutomationDev85 opened a new pull request, #60254:
URL: https://github.com/apache/airflow/pull/60254
# Overview
We observed intermittent hangs in Kubernetes API communication from KubernetesPodOperator, with calls stalling for over a day until tasks were manually stopped. Investigation showed the Kubernetes Python client wasn’t using a client-side timeout, so stalled connections could block indefinitely. To improve robustness, we add a client-side timeout to API calls so they raise a clear exception instead of leaving tasks hanging. This does not fix the underlying cluster/API issue, but it makes failures detectable and recoverable.
We chose a 60-second timeout: long enough to tolerate load, short enough to prevent indefinite hangs. Timeouts are applied per call because there’s no clean, consistent way to set this at client creation across sync/async and watch/exec paths.
# Change Summary
* Set a 60-second client-side timeout for Kubernetes API requests.
* Apply the timeout to individual API calls to ensure stalled calls fail fast.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #60254:
URL: https://github.com/apache/airflow/pull/60254#discussion_r2673941337
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1069,6 +1080,7 @@ async def get_job_status(self, name: str, namespace: str) -> V1Job:
job: V1Job = await v1_api.read_namespaced_job_status(
name=name,
namespace=namespace,
+ **with_timeout(),
Review Comment:
I think adding missing timeout is reasonable - but can you maybe find a way that is on one global place and not scattert across many places where calls are being made? Otherwise I fear this looks very complex and future extensions have the risk missing one or the other place.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #60254:
URL: https://github.com/apache/airflow/pull/60254#issuecomment-3730370361
Overall solid PR. The wrapper class is a great approach.
But there is one potential interaction I wanted to flag based on reading through the code path.. If you look at `def watch_pod_events` (should be line 1005-1054 in your branch), you will see that there is a parameter `timeout_seconds` which has a default argument of 30s (let's call this the 'Watch timeout') . As your new wrapper class `_TimeoutK8sApiClient` (which sets the 'API timeout') is injected when `def get_conn` is called by `def watch_pod_events`.
In idle or stalled watch scenarios (i.e. no events received for a period), this effectively means the API timeout can act as a hard upper bound on how long the watch remains open if it is lower than `timeout_seconds`. In those cases, the watch timeout becomes a best-case limit rather than a guarantee.
Normally, this would not be too concerning but as `def watch_pod_events` is a public API, and this 'API timeout' is an internal default setting, it would not hurt to document this behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on PR #60254:
URL: https://github.com/apache/airflow/pull/60254#issuecomment-3742583599
@SameerMesiah97 I've adopted your text—thanks for the great improvement!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #60254:
URL: https://github.com/apache/airflow/pull/60254#discussion_r2677656646
##########
providers/google/src/airflow/providers/google/cloud/hooks/kubernetes_engine.py:
##########
@@ -498,7 +498,7 @@ def __init__(
)
@contextlib.asynccontextmanager
- async def get_conn(self) -> async_client.ApiClient:
+ async def get_conn(self) -> async_client.ApiClient: # type: ignore[override]
Review Comment:
I assume this change is not needed if comments above are considered.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -239,7 +258,7 @@ def _get_field(self, field_name):
prefixed_name = f"extra__kubernetes__{field_name}"
return self.conn_extras.get(prefixed_name) or None
- def get_conn(self) -> client.ApiClient:
+ def get_conn(self) -> _TimeoutK8sApiClient:
Review Comment:
I'd propose NOT changing the signature of the method as the class passed as return is a private wrapper. Then people looking at type hints are not wondering what this is. As well as the timeout wrapper is not adding any interface of benefit to expose.
```suggestion
def get_conn(self) -> client.ApiClient:
```
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -874,10 +893,10 @@ async def _get_field(self, field_name):
return extras.get(prefixed_name)
@contextlib.asynccontextmanager
- async def get_conn(self) -> async_client.ApiClient:
+ async def get_conn(self) -> AsyncIterator[_TimeoutAsyncK8sApiClient]: # type: ignore[override]
Review Comment:
I assume staying compatible should also prevent the need of the type ignore.
```suggestion
async def get_conn(self) -> async_client.ApiClient:
```
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -307,11 +326,11 @@ def get_conn(self) -> client.ApiClient:
client_configuration=self.client_configuration,
context=cluster_context,
)
- return client.ApiClient()
+ return _TimeoutK8sApiClient()
return self._get_default_client(cluster_context=cluster_context)
- def _get_default_client(self, *, cluster_context: str | None = None) -> client.ApiClient:
+ def _get_default_client(self, *, cluster_context: str | None = None) -> _TimeoutK8sApiClient:
Review Comment:
Same as above: Recommend to keep return declaration compatible
```suggestion
def _get_default_client(self, *, cluster_context: str | None = None) -> client.ApiClient:
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl merged PR #60254:
URL: https://github.com/apache/airflow/pull/60254
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #60254:
URL: https://github.com/apache/airflow/pull/60254#discussion_r2675593986
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1069,6 +1080,7 @@ async def get_job_status(self, name: str, namespace: str) -> V1Job:
job: V1Job = await v1_api.read_namespaced_job_status(
name=name,
namespace=namespace,
+ **with_timeout(),
Review Comment:
@jscheffl Thank you for the response. After reconsidering overnight, I realized that setting a default timeout this way isn’t ideal. I’ve implemented a different approach: a wrapper around both the sync and async API clients that injects a client-side timeout into all API calls.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #60254:
URL: https://github.com/apache/airflow/pull/60254#discussion_r2681481609
##########
providers/google/src/airflow/providers/google/cloud/hooks/kubernetes_engine.py:
##########
@@ -498,7 +498,7 @@ def __init__(
)
@contextlib.asynccontextmanager
- async def get_conn(self) -> async_client.ApiClient:
+ async def get_conn(self) -> async_client.ApiClient: # type: ignore[override]
Review Comment:
done
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -874,10 +893,10 @@ async def _get_field(self, field_name):
return extras.get(prefixed_name)
@contextlib.asynccontextmanager
- async def get_conn(self) -> async_client.ApiClient:
+ async def get_conn(self) -> AsyncIterator[_TimeoutAsyncK8sApiClient]: # type: ignore[override]
Review Comment:
done
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #60254:
URL: https://github.com/apache/airflow/pull/60254#issuecomment-3739977874
> @jscheffl Thank you for your input. This is a good suggestion, and I've implemented the changes accordingly.
>
> @SameerMesiah97 Thank you for the feedback. I've implemented a dynamic approach where the client-side timeout now depends on the server-side timeout. If the server-side timeout exceeds the default API_TIMEOUT, the client-side timeout is set to the server-side timeout plus an offset. This ensures the server has adequate time to respond before the client times out. I believe this is a more robust solution than simply documenting the behavior.
Great. That is even better as it makes the timeout behave more in line with user expectations.
I do not want to sound pedantic here but I think this dynamic approach towards handling conflicting timeouts should be mentioned in the docstring for `def watch_pod_events`. Maybe something like:
`timeout_seconds – Timeout in seconds for the watch stream. A small additional buffer may be applied internally.`
Though the 5 second default is quite short, users scouring the logs might notice the discrepancy and think this is unintentional.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on PR #60254:
URL: https://github.com/apache/airflow/pull/60254#issuecomment-3737658883
@jscheffl Thank you for your input. This is a good suggestion, and I've implemented the changes accordingly.
@SameerMesiah97 Thank you for the feedback. I've implemented a dynamic approach where the client-side timeout now depends on the server-side timeout. If the server-side timeout exceeds the default API_TIMEOUT, the client-side timeout is set to the server-side timeout plus an offset. This ensures the server has adequate time to respond before the client times out. I believe this is a more robust solution than simply documenting the behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KubernetesPodOperator: Fix hanging API calls [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #60254:
URL: https://github.com/apache/airflow/pull/60254#discussion_r2681482577
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -307,11 +326,11 @@ def get_conn(self) -> client.ApiClient:
client_configuration=self.client_configuration,
context=cluster_context,
)
- return client.ApiClient()
+ return _TimeoutK8sApiClient()
return self._get_default_client(cluster_context=cluster_context)
- def _get_default_client(self, *, cluster_context: str | None = None) -> client.ApiClient:
+ def _get_default_client(self, *, cluster_context: str | None = None) -> _TimeoutK8sApiClient:
Review Comment:
done
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -239,7 +258,7 @@ def _get_field(self, field_name):
prefixed_name = f"extra__kubernetes__{field_name}"
return self.conn_extras.get(prefixed_name) or None
- def get_conn(self) -> client.ApiClient:
+ def get_conn(self) -> _TimeoutK8sApiClient:
Review Comment:
done
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org