amoghrajesh commented on code in PR #74010:
URL: https://github.com/apache/airflow/pull/74010#discussion_r4376219073
##########
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:
This part raises a `RuntimeError`, but the `ExecutionAPISecretsBackend`
catches `Exception` and returns `None` at execution_api.py:98, and few other
locations. So on `get_connection` and `get_variable`, sync and async, the new
message is dropped with no log, and the task sees "not found" instead, which is
the same silent wrong answer this PR is trying to stop. Give it a dedicated
class in airflow/sdk/exceptions.py and re-raise it in those four blocks, the
way `AirflowSecretsBackendAccessDenied` already is two lines above each one.
--
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]
