ian-Liaozy commented on code in PR #39161:
URL: https://github.com/apache/beam/pull/39161#discussion_r3553814598
##########
sdks/python/apache_beam/runners/interactive/recording_manager.py:
##########
@@ -820,17 +830,16 @@ def _wait_for_dependencies(
len(computing_deps),
computing_deps.keys())
- futures_to_wait = list(
- set(comp._future for comp in computing_deps.values()))
+ results_to_wait = list(set(comp for comp in computing_deps.values()))
try:
- for i, future in enumerate(futures_to_wait):
+ for i, comp in enumerate(results_to_wait):
if async_result:
async_result.update_display(
- f'Waiting for dependency {i + 1}/{len(futures_to_wait)}...',
- progress=0.05 + 0.05 * (i / len(futures_to_wait)),
+ f'Waiting for dependency {i + 1}/{len(results_to_wait)}...',
+ progress=0.05 + 0.05 * (i / len(results_to_wait)),
)
- future.result()
+ comp.wait_for_completion()
Review Comment:
Hi @claudevdm , now `wait_for_completion()` checks if the future has failed
or been cancelled, and raises the exception directly, and the potential
failures will get caught by the existing `try/except` in
`_wait_for_dependencies()`. Does that looks good to you? Thanks!
--
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]
