chenxuesdu commented on code in PR #37459:
URL: https://github.com/apache/beam/pull/37459#discussion_r2771278211
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java:
##########
@@ -379,4 +393,14 @@ private Timestamp getNextReadChangeStreamEndTimestamp() {
final Timestamp current = Timestamp.now();
return Timestamp.ofTimeSecondsAndNanos(current.getSeconds() + 2 * 60,
current.getNanos());
}
+
+ // For Mutable Change Stream, Spanner only allow the max query end timestamp
to be 2 minutes in
+ // the future.
+ private Timestamp getBoundedQueryEndTimestamp(Timestamp endTimestamp) {
+ if (this.isMutableChangeStream) {
+ Timestamp maxTimestamp = getNextReadChangeStreamEndTimestamp();
Review Comment:
Updated. Please take a look.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java:
##########
@@ -195,13 +200,22 @@ public ProcessContinuation run(
final Timestamp endTimestamp = partition.getEndTimestamp();
final boolean isBoundedRestriction =
!endTimestamp.equals(MAX_INCLUSIVE_END_AT);
final Timestamp changeStreamQueryEndTimestamp =
- isBoundedRestriction ? endTimestamp :
getNextReadChangeStreamEndTimestamp();
+ isBoundedRestriction
+ ? getBoundedQueryEndTimestamp(endTimestamp)
+ : getNextReadChangeStreamEndTimestamp();
// Once the changeStreamQuery completes we may need to resume reading from
the partition if we
// had an unbounded restriction for which we set an arbitrary query end
timestamp and for which
// we didn't encounter any indications that the partition is done
(explicit end records or
// exceptions about being out of timestamp range).
- boolean stopAfterQuerySucceeds = isBoundedRestriction;
+ boolean stopAfterQuerySucceeds = false;
Review Comment:
this is the definition for the boolean.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java:
##########
@@ -195,13 +200,22 @@ public ProcessContinuation run(
final Timestamp endTimestamp = partition.getEndTimestamp();
final boolean isBoundedRestriction =
!endTimestamp.equals(MAX_INCLUSIVE_END_AT);
final Timestamp changeStreamQueryEndTimestamp =
- isBoundedRestriction ? endTimestamp :
getNextReadChangeStreamEndTimestamp();
+ isBoundedRestriction
+ ? getBoundedQueryEndTimestamp(endTimestamp)
+ : getNextReadChangeStreamEndTimestamp();
// Once the changeStreamQuery completes we may need to resume reading from
the partition if we
Review Comment:
updated.
--
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]
