Repository navigation
[Spark 4] Clear the watermark of keys without timers or state in the streaming stateful ParDo - #40550
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40550 +/- ##
============================================
+ Coverage 56.16% 65.03% +8.86%
- Complexity 2288 13947 +11659
============================================
Files 1124 1728 +604
Lines 178336 108384 -69952
Branches 1489 11286 +9797
============================================
- Hits 100170 70492 -29678
+ Misses 75645 33692 -41953
- Partials 2521 4200 +1679
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: @damccorm added as fallback since no labels match configuration 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). |
|
R: @tkaymak would you mind taking a first look at this one? |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
There was a problem hiding this comment.
Thanks @schizophrenicmaniac, the diagnosis is right, and the condition and ordering in storeWatermark hold against persistTimers, reconcileWakeupTimer and Spark deleting the fired timer itself.
The test should prove the cleanup without pinning the late data behavior, see inline.
#40516 adds a test at the same spot in StatefulParDoStreamingTest, please rebase once it merges.
LGTM from my side once the late element and the skew override are gone and the comment states the contract, @Abacn for the merge.
| Set<String> collected = StreamingTestUtils.collected(collectorId); | ||
| assertTrue(collected.contains("k1-0")); | ||
| // The late element finds the key cleared and is filtered like a first visit, see #40446. | ||
| assertTrue(collected.contains("k1-" + LATE_INDEX)); |
There was a problem hiding this comment.
Please drop the late element, LATE_INDEX, the else branch in ExpiryKeySelectorFn and this assertion. It pins the first visit behavior that #40446 wants to tighten, and the zero row assertions already prove the fix, without the change beamWatermark still has one row here. The window expires anyway, the source watermark is declared on the source Dataset and advances on the filtered elements too.
There was a problem hiding this comment.
Done, the late element, LATE_INDEX, the else branch and the assertion are gone. The test now only checks the element got through and that the three state variables have no rows.
| } | ||
|
|
||
| @Override | ||
| public Duration getAllowedTimestampSkew() { |
There was a problem hiding this comment.
getAllowedTimestampSkew is deprecated, with the late element gone this override goes as well.
There was a problem hiding this comment.
Removed.
| private void storeWatermark(SparkTimerInternals timerInternals) { | ||
| checkStateNotNull(beamWatermark).update(timerInternals.currentInputWatermarkTime().getMillis()); | ||
| ValueState<Long> watermarkState = checkStateNotNull(beamWatermark); | ||
| // A key without timers and state, e.g. after all its windows expired, keeps nothing in the |
There was a problem hiding this comment.
Contract only please, for example: A key without timers and state keeps no row, its next visit is a first visit. The reasoning belongs in #40517, not in the code.
There was a problem hiding this comment.
Done, I used your wording.
| } | ||
| } | ||
|
|
||
| private static final class ExpiryTestFn extends DoFn<KV<String, String>, String> { |
There was a problem hiding this comment.
nit: ExpiryTestFn is DownstreamFn with another output string. Reuse DownstreamFn and assert on TestUnboundedSource.element(TAG_EXPIRY, 0, 0).
There was a problem hiding this comment.
Done, switched to DownstreamFn and the assertion is on TestUnboundedSource.element(TAG_EXPIRY, 0, 0) now.
…streaming stateful ParDo
…he contract in storeWatermark
7430a4f to
ce85c06
Compare
4 participants
BeamStatefulProcessor wrote beamWatermark on every visit and never removed it, so each key that ever went through a streaming stateful ParDo kept a row in the state store even after all its windows expired. beamTimers was already cleared by persistTimers once no timers remain. The watermark was the only thing left behind.
storeWatermark, called from handleInputRows and handleExpiredTimer, now clears beamWatermark when the key has no timers and beamState is empty, for example after the GC timer of its last window fired. The change stays inside those two code paths, so it should rebase cleanly onto #40516 and the GroupByKey base-class split.
On late data: after the cleanup, the key's next visit is a first visit and filters late data against TIMESTAMP_MIN_VALUE, so a late element for an already expired window of that key is processed instead of dropped. I went with accepting this as part of #40446. A brand-new key already behaves this way. Keeping any marker per key means keeping the same Long the issue wants gone. A TTLConfig is processing-time based, so it doesn't line up with allowed lateness. I updated the contract comment in timerInternals to cover cleared keys.
The new test, testExpiredKeyLeavesNoState, sends a single element for one key and lets its only window expire as the source watermark advances. It then reads the checkpoint through the statestore data source and asserts that beamState, beamTimers and beamWatermark have no rows left. Without the change one beamWatermark row remains.
Fixes #40517