From 6acc9bdb6f671e791258a061d999137dff665331 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Thu, 15 Oct 2026 16:28:34 +0000 Subject: [PATCH] [Spanner Change Streams] Match single-RPC deadline to total timeout in SpannerIO.ReadChangeStream Set initialRpcTimeout and maxRpcTimeout to 5 minutes to match totalTimeout in SpannerIO.ReadChangeStream.buildChangeStreamSpannerConfig() instead of using a 1-minute per-RPC deadline. --- .../beam/sdk/io/gcp/spanner/SpannerIO.java | 4 +- .../SpannerIOReadChangeStreamTest.java | 52 +++++++++++++++++++ 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java index 4c709963fc68..c5b07715579e 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java @@ -2503,8 +2503,8 @@ SpannerConfig buildChangeStreamSpannerConfig() { .setExecuteStreamingSqlRetrySettings( RetrySettings.newBuilder() .setTotalTimeout(org.threeten.bp.Duration.ofMinutes(5)) - .setInitialRpcTimeout(org.threeten.bp.Duration.ofMinutes(1)) - .setMaxRpcTimeout(org.threeten.bp.Duration.ofMinutes(1)) + .setInitialRpcTimeout(org.threeten.bp.Duration.ofMinutes(5)) + .setMaxRpcTimeout(org.threeten.bp.Duration.ofMinutes(5)) .build()) .build(); } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java index 13fd343f5c83..3c003e5cbd4b 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOReadChangeStreamTest.java @@ -26,6 +26,8 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.google.api.gax.retrying.RetrySettings; +import com.google.api.gax.rpc.StatusCode.Code; import com.google.auth.Credentials; import com.google.cloud.Timestamp; import com.google.cloud.spanner.DatabaseClient; @@ -41,6 +43,7 @@ import org.apache.beam.sdk.extensions.gcp.options.GcpOptions; import org.apache.beam.sdk.io.gcp.spanner.changestreams.MetadataSpannerConfigFactory; import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -175,6 +178,55 @@ public void testSetDirectedReadOptions() { .getReplicaSelections(0) .getLocation()); } + + @Test + public void testDefaultRetrySettings() { + assertNull(spannerConfig.getRetryableCodes()); + assertNull(spannerConfig.getExecuteStreamingSqlRetrySettings()); + SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); + SpannerConfig metadataSpannerConfig = + MetadataSpannerConfigFactory.create( + changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); + ImmutableSet expectedRetryableCodes = ImmutableSet.of(Code.UNAVAILABLE, Code.ABORTED); + RetrySettings expectedRetrySettings = + RetrySettings.newBuilder() + .setTotalTimeout(org.threeten.bp.Duration.ofMinutes(5)) + .setInitialRpcTimeout(org.threeten.bp.Duration.ofMinutes(5)) + .setMaxRpcTimeout(org.threeten.bp.Duration.ofMinutes(5)) + .build(); + assertEquals(expectedRetryableCodes, changeStreamSpannerConfig.getRetryableCodes()); + assertEquals(expectedRetryableCodes, metadataSpannerConfig.getRetryableCodes()); + assertEquals( + expectedRetrySettings, changeStreamSpannerConfig.getExecuteStreamingSqlRetrySettings()); + assertEquals( + expectedRetrySettings, metadataSpannerConfig.getExecuteStreamingSqlRetrySettings()); + } + + @Test + public void testCustomRetrySettings() { + ImmutableSet customRetryableCodes = ImmutableSet.of(Code.UNAVAILABLE); + RetrySettings customRetrySettings = + RetrySettings.newBuilder() + .setTotalTimeout(org.threeten.bp.Duration.ofMinutes(10)) + .setInitialRpcTimeout(org.threeten.bp.Duration.ofMinutes(10)) + .setMaxRpcTimeout(org.threeten.bp.Duration.ofMinutes(10)) + .build(); + spannerConfig = + spannerConfig + .withRetryableCodes(customRetryableCodes) + .withExecuteStreamingSqlRetrySettings(customRetrySettings); + readChangeStream = readChangeStream.withSpannerConfig(spannerConfig); + SpannerConfig changeStreamSpannerConfig = readChangeStream.buildChangeStreamSpannerConfig(); + SpannerConfig metadataSpannerConfig = + MetadataSpannerConfigFactory.create( + changeStreamSpannerConfig, TEST_METADATA_INSTANCE, TEST_METADATA_DATABASE); + assertEquals(customRetryableCodes, changeStreamSpannerConfig.getRetryableCodes()); + assertEquals(customRetryableCodes, metadataSpannerConfig.getRetryableCodes()); + assertEquals( + customRetrySettings, changeStreamSpannerConfig.getExecuteStreamingSqlRetrySettings()); + assertEquals( + customRetrySettings, metadataSpannerConfig.getExecuteStreamingSqlRetrySettings()); + } } /** Parameterized tests for Dialect and Partition Mode combinations. */