hkc-8010 commented on code in PR #74010:
URL: https://github.com/apache/airflow/pull/74010#discussion_r4421076516
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -391,9 +393,18 @@ def _read_frame(self, maxfds: int | None = None) ->
tuple[_ResponseFrame, list[i
return resp, fds or []
return resp
- def _from_frame(self, frame) -> ReceiveMsgType | None:
+ def _from_frame(self, frame, request_id: int | None = None) ->
ReceiveMsgType | None:
from airflow.sdk.exceptions import AirflowRuntimeError
+ # A frame for a different request means the stream is out of step, and
returning it would hand
+ # the caller another request's data.
+ if request_id is not None and frame.id != request_id:
+ raise RuntimeError(
Review Comment:
Done in 7d500f9. Added `AirflowSupervisorCommsOutOfSync` to
`airflow/sdk/exceptions.py` (it subclasses `RuntimeError`, so existing handlers
still match) and re-raised it next to `AirflowSecretsBackendAccessDenied` in
the four blocks in `execution_api.py`.
The dispatchers in `context.py` (`_get_connection`, `_async_get_connection`,
`_get_variable`) have their own `except Exception` that moves on to the next
backend, so I added the same re-raise there. Without it the error was still
dropped one level up.
Tests cover the four backend methods and the three dispatchers. Reverting
the re-raise in either file makes them fail.
--
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]
