damccorm commented on code in PR #32237:
URL: https://github.com/apache/beam/pull/32237#discussion_r1725258052
##########
sdks/python/apache_beam/ml/inference/base.py:
##########
@@ -1305,7 +1346,12 @@ def expand(
return results
def with_exception_handling(
- self, *, exc_class=Exception, use_subprocess=False, threshold=1):
+ self,
+ *,
+ exc_class=Exception,
+ use_subprocess=False,
+ threshold=1,
+ timeout=None):
Review Comment:
I didn't type it to follow the main `with_exception_handling` pattern, but
it probably is still a good idea - updated.
I'll keep the name to match vanilla `with_exception_handling`, updated the
doc string for that piece
##########
sdks/python/apache_beam/ml/inference/base.py:
##########
@@ -1283,13 +1304,33 @@ def expand(
**resource_hints)
if self._with_exception_handling:
+ # On timeouts, report back to the central model metadata
+ # that the model is invalid
+ model_tag = self._model_tag
+ share_across_processes =
self._model_handler.share_model_across_processes(
+ )
+ timeout = self._timeout
+
+ def failure_callback(exception: Exception, element: Any):
+ if type(exception) is not TimeoutError:
+ return
+ model_metadata = load_model_status(model_tag, share_across_processes)
+ model_metadata.try_mark_current_model_invalid(timeout)
+ logging.warning("Operation timed out, etc…….")
Review Comment:
Yikes, yes - that was a miss
--
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]
