github-actions[bot] commented on code in PR #39992:
URL: https://github.com/apache/beam/pull/39992#discussion_r4317493661
##########
sdks/python/apache_beam/runners/worker/bundle_processor.py:
##########
@@ -620,6 +620,17 @@ def __iter__(self) -> Iterator[Any]:
coder_impl.FastPrimitivesCoderImpl.register_iterable_like_type(_ConcatIterable)
+def _await_state_commit(futures) -> None:
+ # Resolve every write before reporting errors, including clears followed by
+ # appends. State futures return error responses rather than raising them.
+ responses = [future.get() for future in futures if future]
+ errors = [
+ response.error for response in responses if response and response.error
+ ]
+ if errors:
+ raise RuntimeError('\n'.join(errors))
Review Comment:
**🟢 Nit**: In `GrpcStateHandler._blocking_request` (`sdk_worker.py`), state
responses with `response.error_reason ==
beam_fn_api_pb2.StateResponse.ErrorReason.CANCELLED` raise
`WorkCancelledException` rather than `RuntimeError` so that
`SdkHarness._execute` logs the cancellation at `INFO` level instead of `ERROR`.
Consider checking `error_reason` here (and propagating it in
`GlobalCachingStateHandler.extend`'s `_DeferredCall`) for consistency with
state read error handling.
--
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]
