[Spanner Change Streams] Fix potential data loss issue by ensuring to only claim timestamps that have been fully processed from the restriction tracker. #37326
Conversation
… only claim timestamps that have been fully processed from the restriction tracker.
Summary of ChangesHello @scwhittle, 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 addresses a critical data integrity issue within the Spanner Change Streams connector. It refines the timestamp claiming and partition completion logic to prevent data loss that could occur when change stream queries reach their end timestamps, especially in cases involving previously unbounded end timestamps. The changes ensure that processing accurately resumes from the last claimed timestamp, enhancing the robustness of continuous data processing. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. 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
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #37326 +/- ##
============================================
+ Coverage 55.16% 57.11% +1.95%
- Complexity 1676 3517 +1841
============================================
Files 1068 1228 +160
Lines 167257 189034 +21777
Branches 1208 3657 +2449
============================================
+ Hits 92261 107964 +15703
- Misses 72816 77653 +4837
- Partials 2180 3417 +1237
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
|
Assigning reviewers: R: @ahmedabu98 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). |
| if (tracker.tryClaim(endTimestamp)) { | ||
| LOG.debug( | ||
| "[{}] change stream completed successfully up to {}", token, changeStreamQueryEndTimestamp); | ||
| if (!tracker.tryClaim(changeStreamQueryEndTimestamp)) { |
There was a problem hiding this comment.
Sam, "An unbounded end timestamp is no longer allowed for the change stream query rpc. However if we reach the end timestamp we should be careful to only advance the tracker to this timestamp and not the possibly unbounded end timestamp of the range." Just a caution to confirm with you:
- The dataflow pipeline can still be configured with bounded endTs and unbounded endTs
- But no matter how the pipeline endTs is configured, here dataflow always break down to use endTs=now()+2m to query spanner change stream.
- We should only claim now()+2m if the query is successful.
| return ProcessContinuation.stop(); | ||
| } | ||
|
|
||
| if (changeStreamQueryEndTimestamp.equals(endTimestamp)) { |
There was a problem hiding this comment.
Should here be >= or just = or no difference?
There was a problem hiding this comment.
This seems not correct if we have partition terminate case. changeStreamQueryEndTimestamp will not be equal to endTimestamp for terminated partition, so the partition will not be marked finished.
|
Thanks @scwhittle ! We are able to reproduce the partition marked finished early, having potential data loss by https://github.com/apache/beam/pull/37339/files (artificial delay the query). We are wondering if the easy fix would be only call getNextReadChangeStreamEndTimestamp immediately before changeStreamQuery, like |
3 participants
An unbounded end timestamp is no longer allowed for the change stream query rpc. However if we reach the end timestamp we should be careful to only advance the tracker to this timestamp and not the possibly unbounded end timestamp of the range.
It is unlikely that the query would end this way due to various timeouts (rpc deadline, interruptor) but it seems that those could possibly be avoided if there were sufficient delays between calculating the query end timestamp and starting the query. A log has been added if this occurs which should only be visible if there would have previously been data loss in this way.
While modifying the file, add a TODO about marking finished as has been discussed.
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.