iralp commented on issue #64628: URL: https://github.com/apache/airflow/issues/64628#issuecomment-6356117444
I'd like to work on this. Before writing code I'd like a maintainer's view on one question — two earlier PRs (#64743, #69508) fixed this the same way and both lapsed without any technical feedback, so there's no decision on record. **What's happening** The supervisor runs a single-threaded `selectors` loop, and the same loop both handles child requests and sends the task heartbeat. `SetXCom` is handled inline, so the loop sits inside one synchronous POST with the whole value held in memory. For a very large value that POST runs longer than `task_instance_heartbeat_timeout` (300s default), no heartbeat goes out while it's in progress, and the scheduler kills the task as a zombie. `execution_api_timeout` (5.0s) doesn't catch it, because httpx timeouts are per-socket-operation rather than per request. **The fix both PRs used, which I think is right** Run the handler on a background thread, keep every socket write to the child on the loop thread, and send the reply once the thread finishes. #69508 is the more complete of the two — it wakes the loop with a socketpair so the reply isn't delayed to the next `select()`, replies strictly in request order, maps both API errors and unexpected exceptions to an `ErrorResponse` so the task is never left waiting, and runs inline on the `dag.test()` path where there's no event loop at all. **The question** Offloading means a huge XCom now *succeeds slowly* instead of failing at 300s. A 300 MB push would hold a task slot for the length of the upload and write a very large `xcom.value` row. There's a less comfortable version of the same point: that 300s heartbeat timeout is currently the only thing bounding this at all. Once heartbeats keep flowing, a push to a server that never finishes responding leaves the task blocked indefinitely — `CommsDecoder._get_response()` is a socket read with no timeout — and nothing in the logs would indicate it. So: should large XComs succeed however long they take, or should there be a size/time ceiling that fails the task with a clear error and points users at `xcom_objectstorage_threshold`? I'm happy to implement either, but they're different PRs, and the second needs a number from someone who knows what's reasonable. **Two things I'd do differently from #69508, if the offload is the wanted direction** 1. It patches the `elif isinstance(msg, ...)` chain, which no longer exists — handlers are now registered through `_register_client_handler`. Making offload a flag on registration (with the wrapper returning `ResponseSent.ALREADY_SENT`) keeps it in one place, and confines it to handlers that only receive `(client, msg)` and therefore can't touch supervisor state. `GetXCom` has the same problem in the download direction, and would then be a one-word change rather than another special case. 2. Both PRs use a `ThreadPoolExecutor` and clean up with `shutdown(wait=False)`. That doesn't cancel the request already in flight, and the pool's threads are non-daemon and joined at interpreter exit — so a task killed mid-upload leaves the supervisor process hanging until the upload finishes, long after there's anyone left to receive the response. The pool's queueing and respawn aren't needed here either, since `CommsDecoder`'s lock means at most one request per child is in flight. A plain `threading.Thread(daemon=True)` holding a standalone `concurrent.futures.Future` keeps the same drain logic and error handling, matches the existing daemon thread at `supervisor.py:2557`, and avoids introducing the first `ThreadPoolExecutor` in `task-sdk`. Happy to take this on if someone can confirm the direction. -- 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]
