damccorm commented on code in PR #41207:
URL: https://github.com/apache/beam/pull/41207#discussion_r4565361948
##########
sdks/python/apache_beam/runners/interactive/interactive_beam_test.py:
##########
@@ -909,6 +909,7 @@ def test_async_computation_result_cancel(self):
# This part is hard to test without deeper runner integration
with self.assertRaises(TimeoutError):
async_result.result(timeout=1) # It should not complete successfully
+ async_result.wait_for_completion(timeout=30)
Review Comment:
Under `InteractiveRunner` (which delegates to `DirectRunner` ->
`FnApiRunner`), `pipeline_to_run.run()` executes the pipeline synchronously
in-process on the `RecordingManager._executor` `ThreadPoolExecutor` worker
thread before `run()` even returns a `PipelineResult`
(`recording_manager.py:491`). Because Python's `ThreadPoolExecutor` /
`threading.Thread` cannot forcibly interrupt a running thread that is already
blocked inside a C-level `time.sleep(100)` (`Future.cancel()` only cancels
futures that haven't started running yet), and because line 911 (`with
self.assertRaises(TimeoutError): async_result.result(timeout=1)`) explicitly
tests that the background thread is still running after `async_result.cancel()`
returns, `cancel()` itself cannot kill `time.sleep(100)`.
However, we can keep the 100s timeout AND immediately unblock the worker
thread during cleanup by replacing `time.sleep(100)` with a `threading.Event`
wait (`InteractiveBeamComputeTest._cancel_event.wait(100)`, using a class
attribute so the `Map` lambda isn't rejected by `PipelineFragment` runner-API
pickling) and calling `InteractiveBeamComputeTest._cancel_event.set()` in a
`finally:` block right before `async_result.wait_for_completion(timeout=30)`. I
have applied and verified this locally on the branch.
--
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]
