Only update watermark if the new value is larger than the existing value - #36456
Only update watermark if the new value is larger than the existing value#36456chenxuesdu wants to merge 5 commits into
Conversation
The updateWatermark is called in QueryChangeStreamAction.updateWatermarkCallback.This funtion update the watermark in a async way, so the update order is not guaranteed. Updated the update watermark logic so that we only move the watermark of one partition forward.
Summary of ChangesHello @chenxuesdu, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request enhances the reliability of watermark management within the Spanner Change Streams connector. By introducing a conditional update mechanism, it guarantees that partition watermarks only advance forward, mitigating issues that could arise from asynchronous updates being processed out of sequence. This ensures data processing integrity and consistent progress tracking. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
| */ | ||
| public Void updateWatermark(String partitionToken, Timestamp watermark) { | ||
| transaction.buffer(createUpdateMetadataWatermarkMutationFrom(partitionToken, watermark)); | ||
| final Struct partition = getPartition(partitionToken); |
There was a problem hiding this comment.
This can be further optimized by only read watermark column, not the whole partition.
partitionToken is the key, maybe have a function getWatermark(partitionToken). Thanks.
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
| transaction.buffer(createUpdateMetadataWatermarkMutationFrom(partitionToken, watermark)); | ||
| final Struct partition = getPartition(partitionToken); | ||
| if (partition == null) { | ||
| LOG.debug("Partiton {} cannot find.", partitionToken); |
| return null; | ||
| } | ||
| final Timestamp currentWatermark = partition.getTimestamp(COLUMN_WATERMARK); | ||
| if (watermark.compareTo(currentWatermark) > 0) { |
There was a problem hiding this comment.
Where does this compare to run (since it's not part of the transaction)? Does it happen atomically along with the read and write? Or are we getting a snapshot, comparing it on the user worker and then writing the value? I'm asking because we need to make sure that if two different processes are trying to update the watermark for the same partition they won't race with each other.
There was a problem hiding this comment.
This compare happens transaction. The function updatewermark is in transaction:
public void updateWatermark(String partitionToken, Timestamp watermark) {
runInTransaction(
transaction -> transaction.updateWatermark(partitionToken, watermark), "updateWatermark");
}
The updateWatermark is called in QueryChangeStreamAction.updateWatermarkCallback.This funtion update the watermark in a async way, so the update order is not guaranteed. Updated the update watermark logic so that we only move the watermark of one partition forward.
dcbdf5b to
3fa50b0
Compare
3fa50b0 to
2e89d80
Compare
4 participants
The updateWatermark is called in
QueryChangeStreamAction.updateWatermarkCallback.This funtion update the watermark in a async way, so the update order is not guaranteed.
Updated the update watermark logic so that we only move the watermark of one partition forward.
Please add a meaningful description for your change here
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.