tianz101 commented on code in PR #37326:
URL: https://github.com/apache/beam/pull/37326#discussion_r2718720993
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java:
##########
@@ -303,14 +303,24 @@ public ProcessContinuation run(
throw e;
}
- LOG.debug("[{}] change stream completed successfully", token);
- if (tracker.tryClaim(endTimestamp)) {
+ LOG.debug(
+ "[{}] change stream completed successfully up to {}", token,
changeStreamQueryEndTimestamp);
+ if (!tracker.tryClaim(changeStreamQueryEndTimestamp)) {
Review Comment:
For v1, old and working behavior is
if bounded endTs, use the bounded endTs.
If unbounded endTs, use the unbounded endTs or now + 2m should both work.
For v2, we have to use now + 2m no matter bounded or unbounded.
Unify v1 and v2 with min(endTs, now + 2m) is nice to have but can be risky
as it changes v1 behavior. So it is really optional for this cl.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
