[damccorm]

Note: this PR was auto-generated by tooling. Before moving it out of draft, the author has done a full review of the PR, specifically evaluating it on the following criteria:

• The PR is an appropriate fix for the issue (e.g. we're not changing a test to avoid a bug)
• The PR is a simple fix to the issue (e.g. we're not adding infra when we could just modify a test)
• The PR is not covering for other problems (e.g. if we're bumping timeouts, it is not because we're covering for a performance problem)

Full details of the methodology of generation and review can be found here - https://docs.google.com/document/d/14lj7ECVov4T4EdwmIzNlWrRGE3lLDMOwX5xUMGzaLoc/edit?usp=sharing

If you have concerns with the PR or with the underlying methodology used to generate this, please comment below - they will be addressed by the author not the tooling :)

Fixes a flaky failure in InteractiveBeamComputeTest.test_compute_blocking (ValueError: All PCollections must belong to the same pipeline.) in PostCommit Python Versions (run 42902232829).

Root Cause

In InteractiveBeamComputeTest.test_async_computation_result_cancel, ib.compute(pcoll, blocking=False) launched a real background thread in RecordingManager._executor running time.sleep(100) on DirectRunner, while the test exited after ~1.1s without waiting for the background thread to finish. Then, the leaked background thread from test_async_computation_result_cancel continued executing during the subsequent test_compute_blocking test and mutated test_compute_blocking's InteractiveEnvironment / pipeline tracker.

Changes

• In InteractiveBeamComputeTest.test_async_computation_result_cancel, shorten the beam.Map sleep from time.sleep(100) to time.sleep(3) and call async_result.wait_for_completion(timeout=30) at the end of the test so the real _execute_pipeline_fragment runs while ensuring no background thread or pipeline execution leaks into subsequent tests.

Verification

• Verified locally with InteractiveBeamComputeTest suite (17/17 passed).
• Triggered beam_PostCommit_Python_Versions.yml workflow run on branch: 43440309029

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR cleanly resolves a test flake in InteractiveBeamComputeTest caused by test_async_computation_result_cancel leaking a 100s background execution thread into subsequent tests. By synchronizing on _execute_pipeline_fragment with threading.Event and ensuring isolated_env.setUp initializes return_value atomically when patching ie.current_env, cross-test state contamination is prevented.

🔍 General Feedback

• The synchronization pattern using started_event and release_event with cleanup in a finally block makes the cancellation test deterministic and eliminates the arbitrary time.sleep delays.
• Guarding self._cancel_button against None in AsyncComputationResult.cancel() properly handles non-IPython environments when IS_IPYTHON is False.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR resolves a flaky test failure in InteractiveBeamComputeTest caused by test_async_computation_result_cancel leaking a 100-second background thread into subsequent tests. The fix cleanly replaces the time.sleep(100) pipeline execution with a synchronized mock of _execute_pipeline_fragment that is properly joined in a finally block, while also closing a brief mock initialization window in isolated_env.setUp and guarding _cancel_button access when IS_IPYTHON is false.

🔍 General Feedback

• Using threading.Event primitives (started_event and release_event) paired with wait_for_completion in finally makes test_async_computation_result_cancel deterministic and eliminates cross-test interference.
• Passing return_value=self.current_env directly to patch(...) in mock_env.py prevents ie.current_env() from briefly returning an unconfigured MagicMock during setUp.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 59.15%. Comparing base (0a963fc) to head (b4f7594).
⚠️ Report is 54 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #41207      +/-   ##
============================================
- Coverage     59.25%   59.15%   -0.10%     
  Complexity     4894     4894              
============================================
  Files          1461     1461              
  Lines        186046   186191     +145     
  Branches       3648     3648              
============================================
- Hits         110244   110146      -98     
- Misses        72445    72688     +243     
  Partials       3357     3357              

Flag	Coverage Δ
python	80.51% <ø> (-0.26%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR resolves a test flake in InteractiveBeamComputeTest where test_async_computation_result_cancel leaked a background pipeline execution thread running time.sleep(100) into subsequent tests. By mocking RecordingManager._execute_pipeline_fragment with a short sleep and awaiting completion before the test exits, the change cleanly eliminates cross-test pollution of the global InteractiveEnvironment.

🔍 General Feedback

• The root cause analysis and fix are spot-on. Mocking _execute_pipeline_fragment and explicitly calling wait_for_completion ensures deterministic cleanup of the background worker thread before isolated_env tears down the test environment.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR fixes a flaky test failure in InteractiveBeamComputeTest where test_async_computation_result_cancel leaked a long-running background pipeline computation (time.sleep(100)) into subsequent tests, mutating their InteractiveEnvironment. Shortening the sleep duration to 3 seconds and awaiting async_result.wait_for_completion(timeout=30) before exiting the test cleanly resolves the background execution leak.

🔍 General Feedback

• The fix is minimal, targeted, and directly addresses the cross-test state contamination while preserving the test's verification of in-flight cancellation behavior.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[damccorm]

Instead of waiting for completion, is there a way to actually force cancel the job? Then we could reraise the time.sleep to 100

[damccorm]

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.