[bvolpato]

AsyncWrapper.teardown() can close the wrapped DoFn's resources while background work still uses them. A bounded DirectRunner pipeline reports DONE, then a delayed write fails with ValueError: I/O operation on closed file.

This change tracks work owned by each wrapper and waits for it before tearing down the wrapped DoFn. Asyncio tasks stay tracked through cancellation cleanup, and the shared executor and event loop remain available to other wrappers.

The AsyncWrapper tests and DirectRunner reproductions passed locally in both thread-pool and asyncio modes. Teardown has no timeout, so hung processing or cleanup can delay shutdown. Pending-output delivery and timer behavior are unchanged.

Run the Python unit suite with:

./gradlew :sdks:python:test-suites:tox:py312:testPython312

---

• If this contribution is large, please file an Apache Individual Contributor License Agreement.
• Replace these instructions with a description of your change and why you have made it.
• Mention the appropriate issue in your description (for example: addresses #123 or fixes #456).
• Update CHANGES.md with noteworthy changes.

See CONTRIBUTING.md for contribution guidelines.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 56.60%. Comparing base (12bbb17) to head (81a0bfd).
⚠️ Report is 19 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #40507       +/-   ##
=============================================
- Coverage     79.83%   56.60%   -23.24%     
- Complexity        0     2288     +2288     
=============================================
  Files           494     1131      +637     
  Lines         83575   179665    +96090     
  Branches          0     1487     +1487     
=============================================
+ Hits          66722   101696    +34974     
- Misses        16853    75448    +58595     
- Partials          0     2521     +2521     

Flag	Coverage Δ
python	80.09% <100.00%> (+0.25%)	⬆️

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.

[github-actions]

📋 Review Summary

This PR fixes a race condition in AsyncWrapper.teardown() where the wrapped DoFn's resources could be torn down while background thread-pool futures or asyncio tasks owned by that wrapper instance were still running or cleaning up after cancellation. The per-instance tracking of futures and asyncio tasks (while keeping the shared executor and event loop alive for other instances) and the accompanying unit and DirectRunner tests are well-structured.

🔍 General Feedback

• Consider guarding teardown() against AsyncWrapper._event_loop being None (e.g., if _sync_fn.setup() fails before the event loop is started, or if reset_state() is called prior to teardown()) and initializing _pending_futures / _pending_async_tasks in __init__ as well as setup().

---

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

[github-actions]

Assigning reviewers:

R: @damccorm for label python.
Backup reviewer: tvalentyn (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

This pull request likely touches a core component ("core" label). Please review with scrutiny.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[damccorm]

@dustin12 would you mind taking a first look at this one?

[damccorm]

R: @dustin12

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers