Repository navigation
Flink dropping watermarks hold fix - #40149
Conversation
|
Assigning reviewers: R: @jrmccluskey 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). |
CarloMariaProietti
changed the title
|
R: @bvolpato |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
damccorm
left a comment
damccorm
left a comment
There was a problem hiding this comment.
Thanks for tracking down and fixing this issue! Left a couple of comments regarding concurrent checkpoints / state reads and a small test nit.
| minBufferedElementTimestamp = | ||
| currentBufferingElementsHandler | ||
| .getElements() | ||
| .mapToLong(e -> e.getTimestamp().getMillis()) | ||
| .min() | ||
| .orElse(Long.MAX_VALUE); |
There was a problem hiding this comment.
A few thoughts on how minBufferedElementTimestamp is recomputed here:
- Concurrent checkpoints (
maxConcurrentCheckpoints > 1): If checkpointNcompletes while checkpointN+1has already been snapshotted (and is still pending innotYetAcknowledgedSnapshots),currentBufferingElementsHandlerhas already rotated to the buffer forN+2. Only checkingcurrentBufferingElementsHandlerwill miss elements still buffered in unacknowledged checkpointN+1and can drop their watermark hold. - Locking (
Locker): When usingKeyedBufferingElementsHandler(e.g., inExecutableStageDoFnOperator),getElements()iterates keys and callsbackend.setCurrentKey(key)/state.get()on theKeyedStateBackend. All other accesses toBufferingElementsHandlerin this class are guarded bytry (Locker lock = locker != null ? locker.get() : null). - Avoiding extra state reads/deserialization: Calling
currentBufferingElementsHandler.getElements()on everycheckpointCompletedforces Flink to read and deserialize all buffered elements in the active buffer (and scan all keys in RocksDB when keyed), only to read and deserialize them again when the next checkpoint completes.
Could we instead track the minimum timestamp per buffer index in memory (e.g., a long[] minTimestampPerBuffer of size numCheckpointBuffers initialized to Long.MAX_VALUE)?
- In
processElement/onTimer, updateminTimestampPerBuffer[currentStateIndex](andminBufferedElementTimestamp). - In
checkpointCompleted, resetminTimestampPerBuffer[toBeAcked.internalId] = Long.MAX_VALUEas each buffer is cleared, and then setminBufferedElementTimestampto the min acrossminTimestampPerBuffer.
That would avoid scanning/deserializing Flink state in checkpointCompleted, avoid the Locker issue, and naturally preserve watermark holds across multiple concurrent checkpoints. (If we also want holds preserved across job restarts, we could populate minTimestampPerBuffer from pendingSnapshots once in initializeState.)
| Supplier<DoFnOperator<String, String, String>> doFnOperatorSupplier = | ||
| () -> | ||
| new DoFnOperator<>( | ||
| doFn, | ||
| "stepName", | ||
| windowedValueCoder, | ||
| Collections.emptyMap(), | ||
| outputTag, | ||
| Collections.emptyList(), | ||
| outputManagerFactory, | ||
| WindowingStrategy.globalDefault(), | ||
| new HashMap<>(), /* side-input mapping */ | ||
| Collections.emptyList(), /* side inputs */ | ||
| options, | ||
| null, | ||
| null, | ||
| DoFnSchemaInformation.create(), | ||
| Collections.emptyMap()); | ||
|
|
||
| DoFnOperator<String, String, String> doFnOperator = doFnOperatorSupplier.get(); | ||
| OneInputStreamOperatorTestHarness<WindowedValue<String>, WindowedValue<String>> testHarness = | ||
| new OneInputStreamOperatorTestHarness<>(doFnOperator); | ||
|
|
||
| testHarness.open(); | ||
|
|
||
| testHarness.processElement( | ||
| new StreamRecord<>(WindowedValues.timestampedValueInGlobalWindow("A", new Instant(10)))); | ||
| testHarness.snapshot(1L, 0L); | ||
| testHarness.processElement( | ||
| new StreamRecord<>(WindowedValues.timestampedValueInGlobalWindow("B", new Instant(20)))); | ||
| testHarness.processWatermark(new Watermark(100)); | ||
| assertThat( | ||
| "A@10 has not yet completed the checkpoint; output watermark must be held at 10 until checkpoint completion", | ||
| doFnOperator.getCurrentOutputWatermark(), | ||
| is(10L)); | ||
|
|
||
| doFnOperator.notifyCheckpointComplete(1L); | ||
| assertThat( | ||
| "A@10 has been checked but B@20 is still buffered; output watermark must not pass it", | ||
| doFnOperator.getCurrentOutputWatermark(), | ||
| is(20L)); | ||
|
|
||
| testHarness.snapshot(2L, 0L); | ||
| doFnOperator.notifyCheckpointComplete(2L); | ||
| assertThat( | ||
| "B@20 has been checked and the buffer is empty; output watermark must advance to input watermark", | ||
| doFnOperator.getCurrentOutputWatermark(), | ||
| is(100L)); | ||
| } |
There was a problem hiding this comment.
Nit: Since this test doesn't restore from a snapshot, doFnOperatorSupplier can be inlined to construct doFnOperator directly, and we should call testHarness.close() at the end (or use try-with-resources). Also might be nice to assert on testHarness.getOutput() at the end to verify that A@10 and B@20 were emitted in order before watermark 100.
|
@damccorm Thank you for reviewing! I've fixed the logic in DoFnOperator so that the minimum timestamp is tracked per checkpoint buffer and extended the test to cover your point. Please let me know whether watermark holds should be preserved across job restarts. |
|
Making sure workflows succeed before proceeding |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #40149 +/- ##
============================================
+ Coverage 56.16% 58.73% +2.56%
- Complexity 2288 13919 +11631
============================================
Files 1124 2585 +1461
Lines 178336 271702 +93366
Branches 1489 11243 +9754
============================================
+ Hits 100170 159593 +59423
- Misses 75645 106107 +30462
- Partials 2521 6002 +3481
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:
|
3 participants
Please add a meaningful description for your change here
Fixes #39991
Problem:
The BufferingDoFnRunner class has the following issue: the minBufferedElementTimestamp field is intended to hold the minimum timestamp among all elements in currentBufferingElementsHandler. This semantic contract can be observed in the processElement and onTimer methods. However, checkpointCompleted violates this contract by setting minBufferedElementTimestamp to Long.MAX_VALUE without guaranteeing that currentBufferingElementsHandler is completely empty (while likely, it is not guaranteed; see #39991 for details).
Solution:
checkpointCompleted now sets minBufferedElementTimestamp to Long.MAX_VALUE only if currentBufferingElementsHandler is empty; otherwise, it recomputes the minimum timestamp according to its semantic invariant.
Implementation Details:
Implemented following the GRASP Polymorphism pattern. The semantics across implementations of the polymorphic getTimestamp method remain consistent with processElement and onTimer.
Testing:
Added a regression test (testExactlyOnceBufferingOutputWatermarkCorrectness) in DoFnOperatorTest that reproduces the issue described in #39991.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
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
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.