bvolpato opened a new pull request, #39996:
URL: https://github.com/apache/beam/pull/39996
`AsyncWrapper` tracks local futures by value (or `id_fn` result), while
runner state and timers are scoped to a key and window. For inputs `('key1',
7)` and `('key2', 7)`, the second key can consume the first key's result and
clear its own state without running its input. Equal identifiers across windows
have the same problem. Timer cleanup can also cancel work belonging to another
window of the same key.
Track futures by `(key, window, identifier)` and pass the window through
scheduling, timer callbacks, and retries. Cleanup now only removes orphaned
work from the current key and window. The persisted bag-state format stays
unchanged.
## 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 42 tests and 4 subtests pass on Python 3.12. The new cases exercise both
the thread-pool and asyncio implementations: equal values and custom IDs across
keys/windows, cleanup preserving another window's pending and completed work,
and timer-driven rescheduling. The four cross-key regression cases fail against
the upstream implementation.
YAPF 0.43.0, Ruff 0.15.22, and `git diff --check` pass for the changed
files. Validation uses the actual wrapper with the test module's existing
state/timer doubles; no end-to-end runner test was added.
## Downsides
Each local tracking entry now includes its key and window. Inputs with
matching identifiers in different keys or windows execute independently and can
occupy separate buffer slots. The configured buffer limit is unchanged.
------------------------
- [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]
