[Python] Finish async bundles after processing outputs

[bvolpato]

The default thread-pool path in AsyncWrapper calls finish_bundle() before iterating the wrapped process() result. Calling a generator function does not execute its body, so a buffered writer can flush an empty buffer and only then enqueue the input. The future completes successfully and the wrapper clears the input from runner state without persisting it.

For a generator-based process() that buffers one row and a non-generator finish_bundle() that flushes the buffer:

Behavior	Lifecycle order	Persisted rows	Remaining runner state
Before	start, finish, process	Empty	Empty
After	start, process, finish	Input row	Empty

Collect the process outputs before calling finish_bundle(). Handle iterable and None results consistently with the asyncio path, including list-valued finish outputs. A generator that raises during iteration now prevents bundle finalization. The existing scalar treatment of strings and bytes is retained.

Reproduction and testing

From sdks/python with the SDK and test dependencies installed:

python -m pytest apache_beam/transforms/async_dofn_test.py -q -n 2

All 38 tests and 6 subtests pass on Python 3.12. New regressions exercise the actual scheduling, future, and commit path with the module's existing state/timer doubles: buffered flushing, list/tuple/iterator process results, list-valued finish output, and a generator exception after yielding. The regressions fail against the upstream thread-pool implementation; asyncio supplies a passing control. No external service is used.

YAPF 0.43.0, Ruff 0.15.22, and git diff --check pass.

Downsides

Non-generator iterable returns now emit their individual elements, matching the DoFn contract and asyncio behavior. Code relying on the previous extra nesting of a returned list or tuple will observe different output shapes.

---

• Describe the bug and include reproducible regression tests.
• Update CHANGES.md with the behavior change.
• Apache Individual Contributor License Agreement, if required for this contribution.

[bvolpato]

Run Python_ML PreCommit 3.12

[bvolpato]

Run Python_ML PreCommit 3.13

[bvolpato]

Run Python_ML PreCommit 3.12

[bvolpato]

Run Python_ML PreCommit 3.12