seanghaeli commented on code in PR #76571:
URL: https://github.com/apache/airflow/pull/76571#discussion_r4496824616
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1181,8 +1184,34 @@ 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 once it gets the request, so only retry
when the request never
+ reached it, and give a slow server longer to answer.
+ """
+ try:
+ Retrying(
+ retry=retry_if_exception_type((httpx2.ConnectError,
httpx2.ConnectTimeout)),
+ stop=stop_after_delay(API_CALLBACK_RUN_RETRY_SECONDS),
+ wait=wait_random_exponential(min=API_RETRY_WAIT_MIN,
max=API_RETRY_WAIT_MAX),
+ before_sleep=_log_and_trace_retry,
+ reraise=True,
+ )(
+ self.client.request,
+ "PATCH",
+ f"callbacks/{callback_id}/run",
+ retry=False,
+ timeout=httpx2.Timeout(API_TIMEOUT,
read=API_CALLBACK_RUN_READ_TIMEOUT),
+ )
+ except (httpx2.ReadTimeout, httpx2.ReadError,
httpx2.RemoteProtocolError) as e:
Review Comment:
Done, it now treats every TransportError except ConnectError and
ConnectTimeout as outcome unknown. WriteError and WriteTimeout were already not
retried, they just missed the log line.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1181,8 +1184,34 @@ 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 once it gets the request, so only retry
when the request never
+ reached it, and give a slow server longer to answer.
+ """
+ try:
+ Retrying(
+ retry=retry_if_exception_type((httpx2.ConnectError,
httpx2.ConnectTimeout)),
+ stop=stop_after_delay(API_CALLBACK_RUN_RETRY_SECONDS),
+ wait=wait_random_exponential(min=API_RETRY_WAIT_MIN,
max=API_RETRY_WAIT_MAX),
+ before_sleep=_log_and_trace_retry,
+ reraise=True,
+ )(
+ self.client.request,
+ "PATCH",
+ f"callbacks/{callback_id}/run",
+ retry=False,
+ timeout=httpx2.Timeout(API_TIMEOUT,
read=API_CALLBACK_RUN_READ_TIMEOUT),
+ )
+ except (httpx2.ReadTimeout, httpx2.ReadError,
httpx2.RemoteProtocolError) as e:
+ log.error(
+ "Callback not started: token exchange outcome unknown, the
server may have used the token, "
+ "so it is not retried",
+ callback_id=callback_id,
+ error_type=type(e).__name__,
+ )
+ raise
Review Comment:
The executor's failure log keeps the exception, but a ReadTimeout reads like
it ran out of retries, not that we chose not to retry because the token may be
used. So, it's not necessarily redundant. I decided to keep one shorter line:
`Callback token exchange outcome unknown, not retrying`.
--
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]
