B-Prod-P9: KeyGroupedInternalPriorityQueue (timer service / windowing)#9
Open
jackylee-ch wants to merge 1 commit into
Open
B-Prod-P9: KeyGroupedInternalPriorityQueue (timer service / windowing)#9jackylee-ch wants to merge 1 commit into
jackylee-ch wants to merge 1 commit into
Conversation
…ue (windowing operators) Spec §6f end-to-end: backend-resident priority queue for Flink 2.2.0's timer service / windowing operators, persisting entries through the ForSt-RS engine instead of an on-heap heap. Composite key encoding (Task 9.1): composite = "q/" || stateName || "/" || kg(2B BE) || ts(8B BE, sign-flipped) || serialize(T) Sign-flip XOR keeps big-endian byte order consistent with signed-numerical order across Long.MIN_VALUE..MAX_VALUE so a prefix scan returns ascending-timestamp entries regardless of sign. Methods (Task 9.2): add, poll, peek, remove, isEmpty, size, addAll, iterator, getSubsetForKeyGroup, removeAll — all routed through linker.put / delete / prefixLookupOpen on the engine's default CF. Tests (Task 9.3): 10 unit tests covering basic add/poll, peek-without-pop, FIFO at same ts, min-heap across distinct ts, sign-flipped negatives, per-key-group isolation, removeAll, size+isEmpty, getSubsetForKeyGroup, and full-range iterator. Backend wiring (Task 9.4): ForStRsAbstractKeyedStateBackend.create(stateName, elementSerializer) (the AbstractKeyedStateBackend override) and the typed public createInternalPriorityQueue(...) helper. Reflective getTimestamp() extraction satisfies the InternalTimer surface natively; explicit ToLongFunction is also exposed for non-timer T types via the queue ctor. ForStRsKeyedStateBackend.getArena() exposed so the queue can share the delegate's FFM arena instead of leaking a parallel one. Tumbling-window IT (Task 9.5): scope reality check applied per Task 9.5 guidance — flink-test-utils not on this module's classpath, so the IT exercises the queue directly (5k events / 500 keys / 5s windows / 8 key groups) instead of standing up a MiniCluster. Same fallback pattern as ForStRsRescalingIT (P4). Production scale tested separately in JMH. EMPTY_VALUE sentinel: 1-byte non-empty marker so engine.get() can distinguish "absent" from "present with empty value" (needed by remove/removeAll bookkeeping). Spotless apply also normalised P8 ForStRsAsyncValueState* — leftover from prior commit's pre-spotless state. Test count: 130/130 unit tests pass (10 new + 120 prior). TumblingWindowIT passes locally. 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
KeyGroupedInternalPriorityQueue<T>for Flink 2.2.0's timer service / windowing operators. Persists entries through the ForSt-RS engine instead of an on-heap heap.q/ || stateName || / || kg(2B BE) || ts(8B BE, sign-flipped) || serialize(T). Sign-flipping the timestamp keeps big-endian byte order consistent with signed-numerical order acrossLong.MIN_VALUE..MAX_VALUE.add/poll/peek/remove/isEmpty/size/addAll/iterator) plusgetSubsetForKeyGroupand a bonusremoveAll(Collection)route throughlinker.put / delete / prefixLookupOpenon the engine's default CF.ForStRsAbstractKeyedStateBackend.create(...)override now constructs the queue; reflectivegetTimestamp()extraction satisfiesInternalTimernatively while a typed lower-level ctor accepts an explicitToLongFunctionfor non-timer T types.Tasks (per plan)
ForStRsKeyGroupedInternalPriorityQueue.java(composite key + interface impl)ForStRsKeyGroupedInternalPriorityQueueTest.java(10 unit tests)ForStRsAbstractKeyedStateBackend.createInternalPriorityQueue(...)TumblingWindowIT.java(scope reality check applied — direct queue exercise; same fallback as P4ForStRsRescalingIT)Adaptations
flink-test-utilsis not on this module's classpath (same as P4 finding). The IT exercises the queue directly (5k events / 500 keys / 5s windows / 8 key-groups) and assertsfired == registered. Production-scale (1M / 100k) is JMH territory — flagged for a future follow-up once a batchedbulkPollFFI lands (per-call iterator-open dominates wall-time at the 1M scale).Keyed<?>bound relaxed on the queue class type-parameter — the production timer type satisfies it but tests can use simpler stand-ins. The Flink-sidecreate(...)override still requires the full bound.EMPTY_VALUEsentinel: 1-byte non-empty value marker soengine.get()can distinguish "absent" from "present-with-empty-value" — needed for accurateremove/removeAllreporting.ForStRsAsyncValueState*(leftover from prior commit).Test plan
mvn test -Dtest='ForStRsKeyGroupedInternalPriorityQueueTest'— 10/10 passmvn test -Dtest='TumblingWindowIT'— passes (5k events / 500 keys, ~minute on JNI test harness)mvn test -Dtest='*Test,!<heavy ITs>,!TumblingWindowIT'— 130/130 pass (10 new + 120 prior)mvn spotless:applycleanmvn compileclean on JDK 25