skymensch commented on code in PR #64743:
URL: https://github.com/apache/airflow/pull/64743#discussion_r3128667987
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -693,6 +704,41 @@ def _cleanup_open_sockets(self):
self.selector.close()
self.stdin.close()
+ self._request_thread_pool.shutdown(wait=False)
+
+ def _drain_pending_requests(self):
+ """Send responses for any offloaded requests that have completed."""
+ remaining: deque[tuple[Future, int]] = deque()
+ while self._pending_requests:
+ future, req_id = self._pending_requests.popleft()
+ if not future.done():
+ remaining.append((future, req_id))
+ continue
+ exc = future.exception()
+ if exc is not None:
+ if isinstance(exc, ServerResponseError):
+ error_details = exc.response.json() if exc.response else
None
+ log.error(
+ "API server error",
+ status_code=exc.response.status_code,
+ detail=error_details,
+ message=str(exc),
+ )
+ self.send_msg(
+ msg=None,
+ error=ErrorResponse(
+ error=ErrorType.API_SERVER_ERROR,
+ detail={
+ "status_code": exc.response.status_code,
+ "message": str(exc),
+ "detail": error_details,
+ },
+ ),
+ request_id=req_id,
+ )
+ else:
+ self.send_msg(msg=None, request_id=req_id)
Review Comment:
Fixed in the latest commit: added an `exc.response is not None` guard before
accessing `.status_code`, and added a fallback `else` branch that sends an
`ErrorResponse` for any exception that is not a `ServerResponseError` (or is a
`ServerResponseError` with a `None` response), so the task process is never
left waiting indefinitely.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -693,6 +704,41 @@ def _cleanup_open_sockets(self):
self.selector.close()
self.stdin.close()
+ self._request_thread_pool.shutdown(wait=False)
+
+ def _drain_pending_requests(self):
+ """Send responses for any offloaded requests that have completed."""
+ remaining: deque[tuple[Future, int]] = deque()
+ while self._pending_requests:
+ future, req_id = self._pending_requests.popleft()
+ if not future.done():
+ remaining.append((future, req_id))
+ continue
+ exc = future.exception()
+ if exc is not None:
+ if isinstance(exc, ServerResponseError):
+ error_details = exc.response.json() if exc.response else
None
Review Comment:
Fixed in the latest commit: wrapped `exc.response.json()` in a `try/except`
block that falls back to `None` if parsing fails, so a malformed or non-JSON
error body cannot bubble out of `_drain_pending_requests()` and disrupt the
supervisor loop.
--
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]
