[Java] Bound the Watch deduplication state with a timestamp cursor - #39746
[Java] Bound the Watch deduplication state with a timestamp cursor#39746Eliaaazzz wants to merge 3 commits into
Conversation
Watch remembers the key of every output it has emitted, so the restriction of an input that is watched indefinitely grows without bound. withTimestampCursor retires a key once the greatest timestamp emitted for that input has moved more than the allowed lateness past it, so the completed set holds a trailing window. Deduplication still goes by output key. An output whose timestamp is below that mark is taken as already seen and is dropped, which suits a poll function whose outputs arrive in roughly non-decreasing timestamp order. A restriction without a cursor keeps the pre-cursor byte format, and a restriction written before the cursor existed retires correctly on its first claim, since the completed set has always recorded each key's output timestamp. Addresses the garbage collection half of apache#18459.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #39746 +/- ##
============================================
+ Coverage 54.97% 57.13% +2.16%
- Complexity 1722 3642 +1920
============================================
Files 1071 1191 +120
Lines 170836 192731 +21895
Branches 1262 3816 +2554
============================================
+ Hits 93910 110120 +16210
- Misses 74656 79046 +4390
- Partials 2270 3565 +1295
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Assigning reviewers: R: @chamikaramj for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Drop the cursor on a round that is not bounding the state, so disabling the option returns the restriction to the pre-cursor encoding and a later round does not resume against a stale floor. Stop on the retention floor rather than the cursor. A cursor at the maximum timestamp still leaves the allowed lateness window claimable, so polling only stops once the floor itself reaches the maximum. Saturate the floor at the minimum timestamp so a large allowed lateness cannot overflow. Document that widening the allowed lateness or dropping the cursor on an update lowers the floor over keys that are already gone, and that a retired key is emitted again only when it returns at or above the current floor.
Abacn
left a comment
Abacn
left a comment
There was a problem hiding this comment.
Thanks! Please check if the same watermark issue in the Python counterpart still applies to here:
If newResults.getOutputs() is empty, and the PollFn did not provide an explicit watermark, computedWatermark remains null and watermark is stale.
| PollingGrowthState<TerminationStateT> pollingRestriction = | ||
| (PollingGrowthState<TerminationStateT>) currentRestriction; | ||
|
|
||
| Duration allowedLateness = spec.getTimestampCursorAllowedLateness(); |
There was a problem hiding this comment.
Consider @Nullable Duration allowedLateness as null has a meaning here.
| return decodeNonPollingGrowthState(is); | ||
| case POLLING_GROWTH_STATE: | ||
| return decodePollingGrowthState(is); | ||
| return decodePollingGrowthState(is, false); |
There was a problem hiding this comment.
This is a good design, make GrowthState coder upgrade compatible (encoded form remains the same for existing NON_POLLING_GROWTH_STATE and POLLING_GROWTH_STATE)
There was a problem hiding this comment.
Added a test pinning the encoded forms byte for byte. I generated the expected bytes by running the same two states through the coder built from master before this branch, and this branch produces identical bytes, so the literals in the test are the pre-cursor encoding. Commit f2f1173.
Nothing below the retention floor is ever emitted, so a round with no new results and no explicit watermark can advance the watermark to the floor instead of leaving it stale. The local for the allowed lateness is annotated Nullable since null keeps the cursor off, and a new coder test pins the pre-cursor encoded forms byte for byte.
|
Checked, and it applied here the same way: an empty round with no explicit watermark left the watermark where it was. With the cursor this is now fixable inside Watch itself. computeNeverSeenBeforeResults drops everything below the retention floor, so no later round can emit below it, and an empty round now advances the watermark to the floor. Without the cursor there is no such bound, so the hash path keeps the conservative behavior and the poll fn stays responsible for the watermark, which MatchPollFn already does with withWatermark(now). Commit f2f1173, with a test that has the watermark stuck at the minimum without the change. |
2 participants
Bounds the deduplication state of the
Watchtransform by event time.Addresses the garbage collection half of #18459, and the TODO left on
PollingGrowthState.getCompleted:What changes
Watchremembers the key of every output it has emitted, so the restriction of an input that is watched indefinitely grows without bound.Growth.withTimestampCursor()adds a cursor toPollingGrowthState, the greatest timestamp emitted for that input. A key is retired fromcompletedonce the cursor has moved more than the allowed lateness past it, so the set holds a trailing window. Deduplication still goes by output key, and a key at the retention floor is kept, so an output that arrives later at the same timestamp is still deduplicated rather than emitted twice.An output whose timestamp is below the floor is taken as already seen and is dropped, since the key that would prove it seen has been retired. That suits a
PollFnwhose outputs arrive in roughly non-decreasing timestamp order, such as one that lists files by last modified time.withTimestampCursor(Duration allowedLateness)widens the window for a source that reports outputs further out of order.The option is off by default and the transform behaves exactly as before without it.
Compatibility
A restriction without a cursor encodes under the existing tag and keeps the pre-cursor bytes, so an in-flight pipeline can be updated onto this version. A restriction written before the cursor existed also retires correctly on its first claim, because
completedhas always recorded each key's own output timestamp rather than a poll time.Known tradeoff
An output that a
PollFnreports again with a later timestamp after its key was retired is emitted a second time. This is documented on the option, and is the reason the option is opt-in.Testing
WatchTest, 28 tests, all passing::sdks:java:core:test --tests "org.apache.beam.sdk.transforms.WatchTest", 20 tests:runners:direct-java:needsRunnerTests --tests "org.apache.beam.sdk.transforms.WatchTest", 8 testsNew coverage: cursor state coder round trip and pre-cursor byte format, key retirement at the cursor, retention widened by allowed lateness, claim rejected below the retention floor, outputs below the floor dropped while an output at the floor is emitted, end to end exactly once delivery with the cursor enabled, and rejection of a negative allowed lateness.
Each new test was checked against a mutated implementation to confirm it fails when the behaviour it covers is removed:
testPollingGrowthTrackerDropsOutputsBehindCursortrySplittestPollingGrowthTrackerRetiresCompletedBehindCursor,testPollingGrowthTrackerAllowedLatenessRetainsCompletedtryClaimtestPollingGrowthTrackerRejectsClaimBehindCursortestMultiplePollsWithTimestampCursorcheckstyleMain,checkstyleTest,spotlessJavaCheck,javadocandspotbugsMainare clean on:sdks:java:core.Relationship to the Python SDK
The same mechanism is under review for the Python
Watchtransform in #39461. The Java API folds the two Python knobs,timestamp_cursorandallowed_lateness, intowithTimestampCursor()andwithTimestampCursor(Duration), so an allowed lateness cannot be set and then silently ignored while the cursor is off. Happy to match the Python surface exactly instead if you would prefer the two SDKs to read the same.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md