This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-1-test by this push:
new 3c66d0674a7 Add configurable timeout for Execution API requests
(#56969)
3c66d0674a7 is described below
commit 3c66d0674a7c1cd6f390e1fece8b8b9fffb63e6c
Author: Kaxil Naik <[email protected]>
AuthorDate: Tue Oct 21 19:58:52 2025 +0100
Add configurable timeout for Execution API requests (#56969)
Workers can now configure the timeout for HTTP requests to the Execution
API server via the `[workers] execution_api_timeout` configuration option.
The default remains 5 seconds (httpx default) to preserve existing behavior.
https://www.python-httpx.org/advanced/timeouts/
This prevents timeout errors in high-load environments where API servers
may take longer than 5 seconds to respond. Users experiencing timeout
issues can now increase this value without code changes.
The timeout controls how long a worker waits for a single API request
to complete, which is different from execution_api_retries that controls
retry behavior after failures.
Fixes #56571
(cherry picked from commit 051160409fb7fd5dd51f27b269734bdb9a1b96d6)
---
airflow-core/src/airflow/config_templates/config.yml | 9 +++++++++
task-sdk/src/airflow/sdk/api/client.py | 5 +++++
task-sdk/tests/task_sdk/api/test_client.py | 19 ++++++++++++++++++-
3 files changed, 32 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 99c5476736a..a7bd0413622 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -1586,6 +1586,15 @@ workers:
type: float
example: ~
default: "90.0"
+ execution_api_timeout:
+ description: |
+ The timeout (in seconds) for HTTP requests from workers to the
Execution API server.
+ This controls how long a worker will wait for a response from the API
server before
+ timing out. Increase this value if you experience timeout errors under
high load.
+ version_added: 3.1.1
+ type: float
+ example: ~
+ default: "5.0"
socket_cleanup_timeout:
description: |
Number of seconds to wait after a task process exits before forcibly
closing any
diff --git a/task-sdk/src/airflow/sdk/api/client.py
b/task-sdk/src/airflow/sdk/api/client.py
index 0a353271c10..6d71a7f106b 100644
--- a/task-sdk/src/airflow/sdk/api/client.py
+++ b/task-sdk/src/airflow/sdk/api/client.py
@@ -810,6 +810,7 @@ API_RETRIES = conf.getint("workers",
"execution_api_retries")
API_RETRY_WAIT_MIN = conf.getfloat("workers", "execution_api_retry_wait_min")
API_RETRY_WAIT_MAX = conf.getfloat("workers", "execution_api_retry_wait_max")
API_SSL_CERT_PATH = conf.get("api", "ssl_cert")
+API_TIMEOUT = conf.getfloat("workers", "execution_api_timeout")
class Client(httpx.Client):
@@ -829,6 +830,10 @@ class Client(httpx.Client):
if API_SSL_CERT_PATH:
ctx.load_verify_locations(API_SSL_CERT_PATH)
kwargs["verify"] = ctx
+
+ # Set timeout if not explicitly provided
+ kwargs.setdefault("timeout", API_TIMEOUT)
+
pyver = f"{'.'.join(map(str, sys.version_info[:3]))}"
super().__init__(
auth=auth,
diff --git a/task-sdk/tests/task_sdk/api/test_client.py
b/task-sdk/tests/task_sdk/api/test_client.py
index cbd7c80bb01..32a709663ac 100644
--- a/task-sdk/tests/task_sdk/api/test_client.py
+++ b/task-sdk/tests/task_sdk/api/test_client.py
@@ -30,7 +30,7 @@ from task_sdk import make_client, make_client_w_dry_run,
make_client_w_responses
from uuid6 import uuid7
from airflow.sdk import timezone
-from airflow.sdk.api.client import RemoteValidationError, ServerResponseError
+from airflow.sdk.api.client import Client, RemoteValidationError,
ServerResponseError
from airflow.sdk.api.datamodels._generated import (
AssetEventsResponse,
AssetResponse,
@@ -99,6 +99,23 @@ class TestClient:
assert isinstance(err.value, FileNotFoundError)
+ @mock.patch("airflow.sdk.api.client.API_TIMEOUT", 60.0)
+ def test_timeout_configuration(self):
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(status_code=200)
+
+ client = make_client(httpx.MockTransport(handle_request))
+ assert client.timeout == httpx.Timeout(60.0)
+
+ def test_timeout_can_be_overridden(self):
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(status_code=200)
+
+ client = Client(
+ base_url="test://server", token="",
transport=httpx.MockTransport(handle_request), timeout=120.0
+ )
+ assert client.timeout == httpx.Timeout(120.0)
+
def test_error_parsing(self):
responses = [
httpx.Response(422, json={"detail": [{"loc": ["#0"], "msg": "err",
"type": "required"}]})
