[feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. #32474
Conversation
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
| } | ||
|
|
||
| continueProcessing &= | ||
| new Instant(timeSupplier.get().toSqlTimestamp()).isBefore(softDeadline) |
There was a problem hiding this comment.
nit: could we do the comparison without having to create a new Instant on every call?
| @@ -88,6 +91,14 @@ public Optional<ProcessContinuation> run( | |||
|
|
|||
| final Timestamp commitTimestamp = record.getCommitTimestamp(); | |||
| final Instant commitInstant = new Instant(commitTimestamp.toSqlTimestamp().getTime()); | |||
| if (tracker instanceof Interruptible | |||
There was a problem hiding this comment.
Could we receive an instance of Interruptible instead of having to do an instanceof comparison on every call?
| @@ -113,6 +114,14 @@ public Optional<ProcessContinuation> run( | |||
|
|
|||
| final Timestamp startTimestamp = record.getStartTimestamp(); | |||
| final Instant startInstant = new Instant(startTimestamp.toSqlTimestamp().getTime()); | |||
| if (tracker instanceof Interruptible | |||
| && !((Interruptible) tracker).shouldContinue(startTimestamp)) { | |||
There was a problem hiding this comment.
I believe this change is to help prevent deadline-exceeds on the change stream rpc from causing problems. The deadlines are in terms of walltime, while the restriction tracker and the record timestamps are in terms of event/data time.
If processing falls behind, the record timestamps and restriction tracker ranges will be in the past but we still want to stop processing after 40 seconds elapses in walltime. Currently the soft-deadline is set to now+40 seconds which might not be reached before the rpc deadline exceeds. And if we changed the soft-deadline to be startTimestamp+40sec, we coudl delay catch up by stopping unnecessarily early if that takes less than 40s of walltime to pass.
Instead I think that you could have a StopWatch at the QueryChangeStreamAction and after 40 seconds of walltime elapses you could update the restriction tracker soft-deadline to the current claimed timestamp. That would ensure that all the records for that timestamp are processed but that newer timestamps would not be claimed.
Currently, Spanner Change Stream IO connector expects the beam framework to issue a split event either 5s or every 5MB of processed payload. In case that such event doesn't occur processing will continue until ultimately 1m pre-set timeout for Change Stream query is reached. This will cause the existing work to be dropped and query to be rescheduled. This feature tracks work for adding a soft deadline before the hard deadline that will allow to commit already processed work and updates internal processing low watermark
addresses #32473 fixes #32473
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.