Repository navigation
Task SDK: send error response on IPC decode failure - #76555
Open
meel-ayush wants to merge 1 commit into
Open
meel-ayush wants to merge 1 commit into
meel-ayush wants to merge 1 commit into
Conversation
When handle_requests() fails to deserialize or validate an incoming request frame, it logged the exception and issued continue. The task runner's recv(4) call was left blocking with nothing ever coming back, deadlocking the child process indefinitely. Meanwhile the supervisor kept heartbeating the API server, so the scheduler's zombie detector never fired — the task would stay stuck in RUNNING until someone killed it manually. Catching the exception and replying with ErrorResponse(GENERIC_ERROR) before continue lets the task runner raise AirflowRuntimeError and exit cleanly.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
When
handle_requests()fails to deserialize or validate an incoming request frame — due to a missing discriminator, schema mismatch, or malformed payload — it was logging the exception and issuingcontinue. No response frame was sent back.The task runner's
CommsDecoder.send()blocks onsocket.recv(4)waiting for the 4-byte length header of the reply. With nothing coming, the child process deadlocks permanently. The supervisor's heartbeat loop keeps running normally, so the scheduler's zombie detector never fires. The task stays stuck inRUNNINGand holds worker capacity until manually killed.The fix catches the exception as
eand replies withErrorResponse(error=ErrorType.GENERIC_ERROR)beforecontinue, matching the existing pattern used forServerResponseErrorand unhandled exceptions later in the same method. The send is wrapped insuppress(Exception)so a secondary socket error during the reply cannot crash the generator loop itself.closes: #76550
Tests
test_handle_requests_decoding_error_sends_error_responseintask-sdk/tests/task_sdk/execution_time/test_supervisor.py, parametrized over an unknown message type and a message missing required fields. Each case asserts the response frame carries aGENERIC_ERROR, the task runner side raisesAirflowRuntimeError, and the generator loop stays alive for the next request.Was generative AI tooling used to co-author this PR?