B-Prod-P8: async state API (Flink 2.x) — spec §6e#8
Open
jackylee-ch wants to merge 1 commit into
Open
Conversation
Per spec §6e: introduces an async-stateful surface for the ForSt-RS
keyed backend. Each public op returns a CompletableFuture<T> and runs
on a virtual-thread executor; per-key serialization is provided by a
shared PerKeyFuturesChain so concurrent ops on the same key are
totally-ordered while ops on distinct keys parallelise freely.
Key components:
- PerKeyFuturesChain<K>: ConcurrentHashMap chain heads with
atomic enqueue + cleanup-on-completion (chain shrinks back to empty).
- ForStRsAsync{Value,List,Map,Reducing,Aggregating}State: thin
wrappers around the sync ForStRsXState classes. The L5 delegate
uses shared mutable buffers (currentKeyBytes / keyOutBuffer) that
are not concurrency-safe; every worker-thread step
(set-key + state-fetch + op) is therefore guarded by
synchronized(backend) — per-key chain still preserves ordering;
the lock just scopes the shared-buffer mutation window.
- ForStRsAsyncValueState additionally exposes an explicit-key API
(value(K) / update(K, T) / clear(K)) so chained continuations
(state.value(k).thenCompose(v -> state.update(k, v + 1))) bind
to the same key on every step without depending on the shared
delegate's currentKey field.
- ForStRsAbstractKeyedStateBackend: 5 new public getters
(getAsyncValueState/ListState/MapState/ReducingState/AggregatingState)
and a lazily-constructed PerKeyFuturesChain backed by
Executors.newVirtualThreadPerTaskExecutor (JDK 21+).
- MapState async iteration returns CompletableFuture<List<...>>
snapshots rather than CompletableFuture<Iterator<...>> because
the underlying FFM iterator + Arena lifetime is owned by the
worker thread; off-thread iterator consumption would be unsafe.
Tests:
- PerKeyFuturesChainTest: 14 unit tests covering ordering,
cross-key parallelism, chain-shrinkage, exception-recovery,
1k-key × 50-op stress.
- ForStRsAsyncValueStateTest: 4 tests including the spec-mandated
100k get-then-put stress on 1k distinct keys, chained per-key
to assert no lost updates (final value per key == 100).
Module test totals: 122 (was 104).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements the async-stateful surface for the ForSt-RS keyed backend (spec §6e).
Key design notes
synchronized(backend). Per-key chain still preserves ordering; the lock just scopes the shared-buffer mutation window. The longer-term path is to either make the delegate thread-safe or shard state objects per key — but neither is required for correctness today.CompletableFuture<List<...>>snapshots rather thanCompletableFuture<Iterator<...>>. The underlying FFM iterator + Arena lifetime is owned by the worker thread; off-thread Iterator consumption would be unsafe.state.value()capturesbackend.getCurrentKey()at call time (caller responsible for no concurrent rebind);state.value(K)accepts the key directly (recommended for chained continuations).Test plan
PerKeyFuturesChainTest— 14 unit tests (12 spec-mandated + 2 bonus): per-key ordering preserved, cross-key parallelism, chain shrinks on completion, exception propagation + chain advances, null/edge cases.ForStRsAsyncValueStateTest— 4 tests including the spec-mandated 100k get-then-put stress on 1k distinct keys; per-key chain serialization is verified by asserting final value == 100 for every key (no lost updates).🤖 Generated with Claude Code