o-nikolas commented on code in PR #76571:
URL: https://github.com/apache/airflow/pull/76571#discussion_r4532402570
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1181,8 +1183,45 @@ def __init__(self, client: Client):
self.client = client
def run(self, callback_id: uuid.UUID) -> None:
- """Exchange the single-use callback token for an execution token."""
- self.client.patch(f"callbacks/{callback_id}/run")
+ """
+ Exchange the single-use callback token for an execution token.
+
+ The server uses up the token on the first request it handles, and a
retry after that gets a 409.
+ So give a slow server longer to answer, and keep retrying for longer
while it is down.
+ """
+ reached_server = False
+
+ def before_sleep(retry_state) -> None:
+ # Only an attempt that may have reached the server can have used
the token.
+ nonlocal reached_server
+ if not isinstance(retry_state.outcome.exception(),
(httpx2.ConnectError, httpx2.ConnectTimeout)):
+ reached_server = True
+ _log_and_trace_retry(retry_state)
+
+ retrying = Retrying(
+ retry=retry_if_exception(_should_retry_api_request),
+ # At least the usual number of attempts, and keep going while the
server is down.
+ stop=stop_after_attempt(API_RETRIES) &
stop_after_delay(API_RETRY_WAIT_MAX),
Review Comment:
Optional: since `execution_api_retry_wait_max` now also sets how long the
callback token exchange keeps retrying, and `execution_api_timeout` ×
`execution_api_retries` sets its read timeout, a sentence in those config
descriptions would help. Otherwise someone tuning them for normal calls won't
know they also change the exchange.
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1229,6 +1268,8 @@ def noop_handler(request: httpx2.Request) ->
httpx2.Response:
API_SSL_CERT_PATH = conf.get("api", "ssl_cert")
API_SSL_CA_FILE_PATH = conf.get("api", "ssl_ca_file", fallback=None)
API_TIMEOUT = conf.getfloat("workers", "execution_api_timeout")
+# The callback token exchange waits for one answer as long as the usual
retries would wait on timeouts.
+API_CALLBACK_RUN_READ_TIMEOUT = API_TIMEOUT * API_RETRIES
Review Comment:
One edge case with basing this on `API_RETRIES`: if someone has `[workers]
execution_api_retries = 0` (which works today and just means a single attempt),
the read timeout here becomes 0. httpx then reads the socket non-blocking, so
the exchange fails right after the server has used the token. I tried it
against a small local server that answers in 50 ms: with retries=0 every
exchange failed (a ReadError, or a 409 after 3-4 requests), while with 1 and 5
it worked first time. Something like `API_TIMEOUT * max(API_RETRIES, 1)` would
avoid it.
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
--
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]
