bvolpato opened a new pull request, #40000: URL: https://github.com/apache/beam/pull/40000
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: ```sh 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. ------------------------ - [x] Describe the bug and include reproducible regression tests. - [x] Update `CHANGES.md` with the behavior change. - [ ] Apache Individual Contributor License Agreement, if required for this contribution. -- 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]
