scwhittle commented on code in PR #40209:
URL: https://github.com/apache/beam/pull/40209#discussion_r4069675218
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java:
##########
@@ -355,12 +363,23 @@ public ProcessContinuation run(
}
LOG.debug(
- "[{}] change stream completed successfully up to {}", token,
changeStreamQueryEndTimestamp);
+ "[{}] change stream completed successfully up to {} (last processed
timestamp: {})",
+ token,
+ changeStreamQueryEndTimestamp,
+ lastProcessedTimestamp);
if (!stopAfterQuerySucceeds) {
- // Records stopped being returned for the query due to our artificial
query end timestamp but
- // we want to continue processing the partition, resuming from
changeStreamQueryEndTimestamp.
- if (!tracker.tryClaim(changeStreamQueryEndTimestamp)) {
+ if (isBoundedRestriction) {
Review Comment:
Actually since stopAfterQuerySucceeds is set to true for bounded
subscriptions going to the end timestamp, I think that we don't need to worry
about the case of never finishing bounded streams here.
I think that this could just be
LOG.debug(...)
if (!stopAfterQuerySucceeds) {
// Leave the tracker at the last claimed position (record or heartbeat)
// instead of advancing to the query end timestamp.
// This works around spanner backend issue where some child partition
records
// were not sent. In other cases since heartbeating is regular the last
// received timestamp will not be far behind the query end timestamp.
return ProcessContinuation.resume();
}
bundleFinalizer.afterBundleCommit(...);
--
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]
