[Python] Isolate async work by key and window

[bvolpato]

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 (encoded_key, window, identifier) and pass the window through scheduling, timer callbacks, and retries. The key uses the existing bag-state key coder: its encoded identity survives state decoding, including NaN keys, and distinguishes numeric keys that compare equal in Python, such as -0.0 and 0.0. Cleanup only removes orphaned work from the current encoded key and window. The persisted bag-state format stays unchanged.

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 48 tests and 10 subtests pass on Python 3.12.

The 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. Additional state-coder roundtrip cases cover NaN completion, signed-zero and mixed numeric-key separation, and cancellation preserving another encoded key's work. Those cases fail with raw Python keys in the tracking map.

YAPF 0.43.0, Ruff 0.15.22, and git diff --check pass for the changed files. Validation uses the actual wrapper with state/timer doubles, including the real bag-state coder for the numeric-key cases; no end-to-end runner test was added.

Downsides

Each local tracking entry now includes encoded key bytes and its window. Inputs with matching identifiers in different keys or windows execute independently and can occupy separate buffer slots. The configured buffer limit is unchanged.

---

• 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.10