You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "dedocibula (via GitHub)" <gi...@apache.org> on 2024/09/16 19:34:14 UTC
[PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
dedocibula opened a new pull request, #32474:
URL: https://github.com/apache/beam/pull/32474
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:
- [ ] Mention the appropriate issue in your description (for example: `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, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878946547
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ChildPartitionsRecordAction.java:
##########
@@ -105,6 +106,7 @@ public Optional<ProcessContinuation> run(
PartitionMetadata partition,
ChildPartitionsRecord record,
RestrictionTracker<TimestampRange, Timestamp> tracker,
+ RestrictionInterrupter<Timestamp> interrupter,
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878908551
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupterTest.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Test;
+
+public class RestrictionInterrupterTest {
+
+ @Test
+ public void testTryInterrupt() {
+ RestrictionInterrupter<Integer> interrupter =
+ new RestrictionInterrupter<Integer>(
+ () -> Instant.ofEpochSecond(0), Duration.standardSeconds(30));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(10));
+ assertFalse(interrupter.tryInterrupt(1));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(15));
+ assertFalse(interrupter.tryInterrupt(2));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(30));
+ assertFalse(interrupter.tryInterrupt(3));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(40));
+ assertFalse(interrupter.tryInterrupt(3));
Review Comment:
Added comment
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupterTest.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Test;
+
+public class RestrictionInterrupterTest {
+
+ @Test
+ public void testTryInterrupt() {
+ RestrictionInterrupter<Integer> interrupter =
+ new RestrictionInterrupter<Integer>(
+ () -> Instant.ofEpochSecond(0), Duration.standardSeconds(30));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(10));
+ assertFalse(interrupter.tryInterrupt(1));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(15));
+ assertFalse(interrupter.tryInterrupt(2));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(30));
+ assertFalse(interrupter.tryInterrupt(3));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(40));
+ assertFalse(interrupter.tryInterrupt(3));
+ assertTrue(interrupter.tryInterrupt(4));
+ assertTrue(interrupter.tryInterrupt(5));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(50));
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32474:
URL: https://github.com/apache/beam/pull/32474#issuecomment-2353766882
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "thiagotnunes (via GitHub)" <gi...@apache.org>.
thiagotnunes commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1762261479
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/ReadChangeStreamPartitionRangeTracker.java:
##########
@@ -49,6 +54,48 @@ public ReadChangeStreamPartitionRangeTracker(PartitionMetadata partition, Timest
this.partition = partition;
}
+ /**
+ * Sets a soft timeout from now for processing new positions. After the timeout the shouldContinue
+ * will start returning false indicating an early exit from processing.
+ */
+ @Override
+ public void setSoftTimeout(Duration duration) {
+ softDeadline = new Instant(timeSupplier.get().toSqlTimestamp()).plus(duration);
+ continueProcessing = true;
+ }
+
+ /**
+ * Returns true if the restriction tracker can claim new positions.
+ *
+ * <p>If soft timeout isn't set always returns true. Otherwise:
+ *
+ * <ol>
+ * <li>If soft deadline hasn't been reached always returns true.
+ * <li>If soft deadline has been reached but we haven't processed any positions returns true.
+ * <li>If soft deadline has been reached but the new position is the same as the last attempted
+ * position returns true.
+ * <li>If soft deadline has been reached and the new position differs from the last attempted
+ * position returns false.
+ * </ol>
+ *
+ * @return {@code true} if the position processing should continue, {@code false} if the soft
+ * deadline has been reached and we have fully processed the previous position.
+ */
+ @Override
+ public boolean shouldContinue(Timestamp position) {
+ if (!continueProcessing) {
+ return false;
+ }
+ if (softDeadline == null || lastAttemptedPosition == null) {
+ return true;
+ }
+
+ continueProcessing &=
+ new Instant(timeSupplier.get().toSqlTimestamp()).isBefore(softDeadline)
Review Comment:
nit: could we do the comparison without having to create a `new Instant` on every call?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/DataChangeRecordAction.java:
##########
@@ -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
Review Comment:
Could we receive an instance of Interruptible instead of having to do an instanceof comparison on every call?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #32474:
URL: https://github.com/apache/beam/pull/32474
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878914793
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Ack
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878904575
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java:
##########
@@ -62,6 +63,7 @@ public class QueryChangeStreamAction {
private static final Logger LOG = LoggerFactory.getLogger(QueryChangeStreamAction.class);
private static final Duration BUNDLE_FINALIZER_TIMEOUT = Duration.standardMinutes(5);
+ private static final Duration RESTRICTION_TRACKER_TIMEOUT = Duration.standardSeconds(40);
Review Comment:
Added comment. Basically we are trying to select a value lower than the predetermined deadline - https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java#L1859-L1871
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on PR #32474:
URL: https://github.com/apache/beam/pull/32474#issuecomment-2353764367
R: @thiagotnunes @scwhittle
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878911241
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class RestrictionInterrupter<T> {
+ private T lastAttemptedPosition;
+
+ private Supplier<Instant> timeSupplier;
+ private Instant softDeadline;
+ private boolean hasInterrupted = true;
+
+ /**
+ * Sets a soft timeout from now for processing new positions. After the timeout the tryInterrupt
+ * will start returning true indicating an early exit from processing.
+ */
+ public static <T> RestrictionInterrupter<T> withSoftTimeout(Duration timeout) {
+ return new RestrictionInterrupter<T>(() -> Instant.now(), timeout);
+ }
+
+ RestrictionInterrupter(Supplier<Instant> timeSupplier, Duration timeout) {
+ this.timeSupplier = timeSupplier;
+ this.softDeadline = this.timeSupplier.get().plus(timeout);
+ hasInterrupted = false;
+ }
+
+ @VisibleForTesting
+ void setTimeSupplier(Supplier<Instant> timeSupplier) {
+ this.timeSupplier = timeSupplier;
+ }
+
+ /**
+ * Returns true if the restriction tracker should be interrupted in claiming new positions.
+ *
+ * <ol>
+ * <li>If soft deadline hasn't been reached always returns false.
+ * <li>If soft deadline has been reached but we haven't processed any positions returns false.
+ * <li>If soft deadline has been reached but the new position is the same as the last attempted
+ * position returns false.
+ * <li>If soft deadline has been reached and the new position differs from the last attempted
+ * position returns true.
+ * </ol>
+ *
+ * @return {@code true} if the position processing should continue, {@code false} if the soft
+ * deadline has been reached and we have fully processed the previous position.
+ */
+ public boolean tryInterrupt(T position) {
+ if (hasInterrupted) {
+ return true;
+ }
+ if (lastAttemptedPosition == null) {
+ lastAttemptedPosition = position;
+ return false;
+ }
+
+ hasInterrupted |=
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878920234
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ChildPartitionsRecordAction.java:
##########
@@ -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)) {
Review Comment:
Discussed offline. That's what the current logic does
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "thiagotnunes (via GitHub)" <gi...@apache.org>.
thiagotnunes commented on PR #32474:
URL: https://github.com/apache/beam/pull/32474#issuecomment-2574312508
No new comments from me, I think Andrew addressed the ones I had before.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878912153
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class RestrictionInterrupter<T> {
+ private T lastAttemptedPosition;
+
+ private Supplier<Instant> timeSupplier;
+ private Instant softDeadline;
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878156969
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java:
##########
@@ -62,6 +63,7 @@ public class QueryChangeStreamAction {
private static final Logger LOG = LoggerFactory.getLogger(QueryChangeStreamAction.class);
private static final Duration BUNDLE_FINALIZER_TIMEOUT = Duration.standardMinutes(5);
+ private static final Duration RESTRICTION_TRACKER_TIMEOUT = Duration.standardSeconds(40);
Review Comment:
Can you comment where this timeout comes from? Ie what is the rpc deadline we are trying to avoid hitting that can currently cause an exception? If the deadline is O(minutes) we might want to increase this to give the dofn the ability to catch up more if it is behind.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
can you remove this and add Nullable annotations where needed?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class RestrictionInterrupter<T> {
+ private T lastAttemptedPosition;
+
+ private Supplier<Instant> timeSupplier;
+ private Instant softDeadline;
Review Comment:
final
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupterTest.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Test;
+
+public class RestrictionInterrupterTest {
+
+ @Test
+ public void testTryInterrupt() {
+ RestrictionInterrupter<Integer> interrupter =
+ new RestrictionInterrupter<Integer>(
+ () -> Instant.ofEpochSecond(0), Duration.standardSeconds(30));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(10));
+ assertFalse(interrupter.tryInterrupt(1));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(15));
+ assertFalse(interrupter.tryInterrupt(2));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(30));
+ assertFalse(interrupter.tryInterrupt(3));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(40));
+ assertFalse(interrupter.tryInterrupt(3));
Review Comment:
add a comment to help future test readers
// Though the deadline has passed same position as previously accepted is not interrupted.
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupterTest.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Test;
+
+public class RestrictionInterrupterTest {
+
+ @Test
+ public void testTryInterrupt() {
+ RestrictionInterrupter<Integer> interrupter =
+ new RestrictionInterrupter<Integer>(
+ () -> Instant.ofEpochSecond(0), Duration.standardSeconds(30));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(10));
+ assertFalse(interrupter.tryInterrupt(1));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(15));
+ assertFalse(interrupter.tryInterrupt(2));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(30));
+ assertFalse(interrupter.tryInterrupt(3));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(40));
+ assertFalse(interrupter.tryInterrupt(3));
+ assertTrue(interrupter.tryInterrupt(4));
+ assertTrue(interrupter.tryInterrupt(5));
+ interrupter.setTimeSupplier(() -> Instant.ofEpochSecond(50));
Review Comment:
could verify that even if the clock is non-monotonic that once true is returned it always returns true
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class RestrictionInterrupter<T> {
+ private T lastAttemptedPosition;
+
+ private Supplier<Instant> timeSupplier;
+ private Instant softDeadline;
+ private boolean hasInterrupted = true;
+
+ /**
+ * Sets a soft timeout from now for processing new positions. After the timeout the tryInterrupt
+ * will start returning true indicating an early exit from processing.
+ */
+ public static <T> RestrictionInterrupter<T> withSoftTimeout(Duration timeout) {
+ return new RestrictionInterrupter<T>(() -> Instant.now(), timeout);
+ }
+
+ RestrictionInterrupter(Supplier<Instant> timeSupplier, Duration timeout) {
+ this.timeSupplier = timeSupplier;
+ this.softDeadline = this.timeSupplier.get().plus(timeout);
+ hasInterrupted = false;
+ }
+
+ @VisibleForTesting
+ void setTimeSupplier(Supplier<Instant> timeSupplier) {
+ this.timeSupplier = timeSupplier;
+ }
+
+ /**
+ * Returns true if the restriction tracker should be interrupted in claiming new positions.
+ *
+ * <ol>
+ * <li>If soft deadline hasn't been reached always returns false.
+ * <li>If soft deadline has been reached but we haven't processed any positions returns false.
+ * <li>If soft deadline has been reached but the new position is the same as the last attempted
+ * position returns false.
+ * <li>If soft deadline has been reached and the new position differs from the last attempted
+ * position returns true.
+ * </ol>
+ *
+ * @return {@code true} if the position processing should continue, {@code false} if the soft
+ * deadline has been reached and we have fully processed the previous position.
+ */
+ public boolean tryInterrupt(T position) {
+ if (hasInterrupted) {
+ return true;
+ }
+ if (lastAttemptedPosition == null) {
+ lastAttemptedPosition = position;
+ return false;
+ }
+
+ hasInterrupted |=
Review Comment:
nit: seems clearer flow to check position first separately
if (position.equals(lastAttemptedPosition)) return false;
lastAttemptedPosition = position;
hasInterrupted |= timeSupplier.get().isAfter(softDeadline)
return hasInterrupted;
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ChildPartitionsRecordAction.java:
##########
@@ -105,6 +106,7 @@ public Optional<ProcessContinuation> run(
PartitionMetadata partition,
ChildPartitionsRecord record,
RestrictionTracker<TimestampRange, Timestamp> tracker,
+ RestrictionInterrupter<Timestamp> interrupter,
Review Comment:
add to comment
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878925343
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/DataChangeRecordAction.java:
##########
@@ -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
Review Comment:
Added as explicit class as this route proved to be infeasible due to Beam adapter pattern
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #32474:
URL: https://github.com/apache/beam/pull/32474#issuecomment-2535278162
@thiagotnunes Any other comments before merging?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1770986607
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ChildPartitionsRecordAction.java:
##########
@@ -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)) {
Review 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.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #32474:
URL: https://github.com/apache/beam/pull/32474#issuecomment-2456662744
Friendly ping on this, as it would be nice to avoid deadline exceeds from causing retries
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #32474:
URL: https://github.com/apache/beam/pull/32474#issuecomment-2572621115
Rerunning tests and will merge if green, sorry for the delay I lost track of this before going OOO.
@thiagotnunes another ping in case you had some comments
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle closed pull request #32474: [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector.
URL: https://github.com/apache/beam/pull/32474
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on PR #32474:
URL: https://github.com/apache/beam/pull/32474#issuecomment-2533001909
Tested with BigQuery template and sdf checkpoint interval = 80s. There don't seem to be deadline exceeded anymore and data freshness by state oscillates between 40-50s
gcloud dataflow flex-template run $DATAFLOW_JOB --template-file-gcs-location gs://$BUCKET_NAME/templates/flex/Spanner_Change_Streams_to_BigQuery --region us-central1 --parameters spannerInstanceId=$INSTANCE --parameters spannerDatabase=$DATABASE --parameters spannerMetadataInstanceId=$INSTANCE --parameters spannerMetadataDatabase=$META_DATABASE --parameters spannerChangeStreamName=$CHANGE_STREAM --parameters bigQueryDataset=$BIG_QUERY_DATASET --parameters deadLetterQueueDirectory=gs://$BUCKET_NAME/dlq/try1 --parameters experiments=sdf_checkpoint_after_duration=80s
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878947564
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Added
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class RestrictionInterrupter<T> {
+ private T lastAttemptedPosition;
+
+ private Supplier<Instant> timeSupplier;
+ private Instant softDeadline;
+ private boolean hasInterrupted = true;
+
+ /**
+ * Sets a soft timeout from now for processing new positions. After the timeout the tryInterrupt
+ * will start returning true indicating an early exit from processing.
+ */
+ public static <T> RestrictionInterrupter<T> withSoftTimeout(Duration timeout) {
+ return new RestrictionInterrupter<T>(() -> Instant.now(), timeout);
+ }
+
+ RestrictionInterrupter(Supplier<Instant> timeSupplier, Duration timeout) {
+ this.timeSupplier = timeSupplier;
+ this.softDeadline = this.timeSupplier.get().plus(timeout);
+ hasInterrupted = false;
+ }
+
+ @VisibleForTesting
+ void setTimeSupplier(Supplier<Instant> timeSupplier) {
+ this.timeSupplier = timeSupplier;
+ }
+
+ /**
+ * Returns true if the restriction tracker should be interrupted in claiming new positions.
+ *
+ * <ol>
+ * <li>If soft deadline hasn't been reached always returns false.
+ * <li>If soft deadline has been reached but we haven't processed any positions returns false.
+ * <li>If soft deadline has been reached but the new position is the same as the last attempted
+ * position returns false.
+ * <li>If soft deadline has been reached and the new position differs from the last attempted
+ * position returns true.
+ * </ol>
+ *
+ * @return {@code true} if the position processing should continue, {@code false} if the soft
+ * deadline has been reached and we have fully processed the previous position.
+ */
+ public boolean tryInterrupt(T position) {
+ if (hasInterrupted) {
+ return true;
+ }
+ if (lastAttemptedPosition == null) {
+ lastAttemptedPosition = position;
+ return false;
+ }
+
+ hasInterrupted |=
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat 32473] Added soft deadline logic to Spanner Change Stream IO connector. [beam]
Posted by "dedocibula (via GitHub)" <gi...@apache.org>.
dedocibula commented on code in PR #32474:
URL: https://github.com/apache/beam/pull/32474#discussion_r1878947564
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Added
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/RestrictionInterrupter.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction;
+
+import java.util.function.Supplier;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An interrupter for restriction tracker of type T. */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class RestrictionInterrupter<T> {
+ private T lastAttemptedPosition;
+
+ private Supplier<Instant> timeSupplier;
+ private Instant softDeadline;
+ private boolean hasInterrupted = true;
+
+ /**
+ * Sets a soft timeout from now for processing new positions. After the timeout the tryInterrupt
+ * will start returning true indicating an early exit from processing.
+ */
+ public static <T> RestrictionInterrupter<T> withSoftTimeout(Duration timeout) {
+ return new RestrictionInterrupter<T>(() -> Instant.now(), timeout);
+ }
+
+ RestrictionInterrupter(Supplier<Instant> timeSupplier, Duration timeout) {
+ this.timeSupplier = timeSupplier;
+ this.softDeadline = this.timeSupplier.get().plus(timeout);
+ hasInterrupted = false;
+ }
+
+ @VisibleForTesting
+ void setTimeSupplier(Supplier<Instant> timeSupplier) {
+ this.timeSupplier = timeSupplier;
+ }
+
+ /**
+ * Returns true if the restriction tracker should be interrupted in claiming new positions.
+ *
+ * <ol>
+ * <li>If soft deadline hasn't been reached always returns false.
+ * <li>If soft deadline has been reached but we haven't processed any positions returns false.
+ * <li>If soft deadline has been reached but the new position is the same as the last attempted
+ * position returns false.
+ * <li>If soft deadline has been reached and the new position differs from the last attempted
+ * position returns true.
+ * </ol>
+ *
+ * @return {@code true} if the position processing should continue, {@code false} if the soft
+ * deadline has been reached and we have fully processed the previous position.
+ */
+ public boolean tryInterrupt(T position) {
+ if (hasInterrupted) {
+ return true;
+ }
+ if (lastAttemptedPosition == null) {
+ lastAttemptedPosition = position;
+ return false;
+ }
+
+ hasInterrupted |=
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org