This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f8c7ead5b4a Task SDK: send error response on IPC decode failure
(#76555)
f8c7ead5b4a is described below
commit f8c7ead5b4a3af36c9c6706f122830a351cb7cc3
Author: Ayush Meel <[email protected]>
AuthorDate: Fri Nov 13 19:33:24 2026 +0530
Task SDK: send error response on IPC decode failure (#76555)
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(GEN [...]
---
.../src/airflow/sdk/execution_time/supervisor.py | 15 +++++++-
.../task_sdk/execution_time/test_supervisor.py | 42 ++++++++++++++++++++++
2 files changed, 56 insertions(+), 1 deletion(-)
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 4ab7889603b..6fd90a0785c 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -1179,8 +1179,21 @@ class WatchedSubprocess:
try:
msg =
self.decoder.validate_python(self._deserialize_request(request.body))
- except Exception:
+ except Exception as e:
log.exception("Unable to decode message", body=request.body)
+ with suppress(Exception):
+ self.send_msg(
+ msg=None,
+ error=ErrorResponse(
+ error=ErrorType.GENERIC_ERROR,
+ detail={
+ "status_code": None,
+ "message": f"Unable to decode message: {e}",
+ "exception_type": type(e).__name__,
+ },
+ ),
+ request_id=request.id,
+ )
continue
# Restore the task runner's trace context so that any outbound
HTTP calls made while
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 8e986595318..d9d1db6587d 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -4708,6 +4708,48 @@ class TestHandleRequest:
# Should not raise StopIteration (which would mean the loop crashed).
generator.send(req2)
+ @pytest.mark.parametrize(
+ "invalid_body",
+ [
+ pytest.param({"type": "NonExistentMessageType"},
id="unknown_message_type"),
+ pytest.param({"type": "GetVariable"}, id="missing_required_field"),
+ ],
+ )
+ def test_handle_requests_decoding_error_sends_error_response(
+ self, watched_subprocess, mocker, invalid_body
+ ):
+ """Test that invalid request frames produce an error response rather
than dropping the frame."""
+ watched_subprocess, read_socket = watched_subprocess
+
+ generator = watched_subprocess.handle_requests(log=mocker.Mock())
+ next(generator)
+
+ req_frame = _RequestFrame(id=randint(1, 2**32 - 1), body=invalid_body)
+ generator.send(req_frame)
+
+ read_socket.settimeout(0.5)
+ frame_len = int.from_bytes(read_socket.recv(4), "big")
+ bytes_ = read_socket.recv(frame_len)
+ frame = msgspec.msgpack.Decoder(_ResponseFrame).decode(bytes_)
+
+ assert frame.id == req_frame.id
+ assert frame.error is not None
+ assert frame.error["error"] == "GENERIC_ERROR"
+ assert "Unable to decode message" in frame.error["detail"]["message"]
+
+ comms = CommsDecoder(socket=None)
+ with pytest.raises(AirflowRuntimeError) as exc_info:
+ comms._from_frame(frame)
+
+ assert exc_info.value.error.error == ErrorType.GENERIC_ERROR
+ assert exc_info.value.error.detail["exception_type"] is not None
+
+ watched_subprocess.client.task_instances.succeed =
mocker.Mock(return_value=None)
+ msg = SucceedTask(end_date=timezone.parse("2024-10-31T12:00:00Z"))
+ req_valid = _RequestFrame(id=randint(1, 2**32 - 1),
body=msg.model_dump())
+ generator.send(req_valid)
+ watched_subprocess.client.task_instances.succeed.assert_called_once()
+
@pytest.mark.parametrize(
("msg", "api_method", "expected_state"),
[
