From 1c7b16ac25d86065f5d7d0c44172a7e318ef773f Mon Sep 17 00:00:00 2001 From: Rahul Yadav Date: Fri, 11 Sep 2026 10:59:55 +0530 Subject: [PATCH 1/4] fix(spanner): honor maxAttempts and totalTimeout in streaming resume loop --- .../spanner/ResumableStreamIterator.java | 133 ++++++- .../cloud/spanner/MockSpannerServiceImpl.java | 13 + .../spanner/ResumableStreamIteratorTest.java | 192 +++++++++- .../StreamingRetryBudgetMockServerTest.java | 357 ++++++++++++++++++ 4 files changed, 670 insertions(+), 25 deletions(-) create mode 100644 java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java index aac7f63c8614..88a01114e0a3 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java @@ -52,6 +52,23 @@ * track of the most recent resume token seen, and will buffer partial result set chunks that do not * have a resume token until one is seen or buffer space is exceeded, which reduces the chance of * yielding data to the caller that cannot be resumed. + * + *

When non-default streaming retry settings are used, the resume loop is bounded by those + * settings: consecutive failed attempts are limited by {@link RetrySettings#getMaxAttempts()} if + * that has been set to a value greater than zero, and the total time spent on a sequence of + * consecutive failed attempts is limited by {@link RetrySettings#getTotalTimeout()} if that has + * been set to a positive value. Only consecutive failures count against these budgets: any + * progress on the stream (that is, receiving a resume token that differs from the last seen resume + * token) resets both, so a long-running stream that regularly makes progress is not terminated by + * an occasional transient error. + * + *

These limits bound the number of streams that this iterator starts. Each (re)started stream + * is a call through the underlying GAX callable, and GAX applies the same {@link RetrySettings} to + * attempts of that call that fail before any response has been received. A stream that repeatedly + * fails before its first response can therefore consist of up to maxAttempts RPC attempts itself, + * so a configured maxAttempts of N bounds the total number of RPC attempts without progress by + * N*N, not by N. The total timeout is measured in wall-clock time from the first failure of the + * sequence and therefore spans both layers. */ @VisibleForTesting abstract class ResumableStreamIterator extends AbstractIterator @@ -69,7 +86,21 @@ abstract class ResumableStreamIterator extends AbstractIterator stream; + + /** + * The number of consecutive failed attempts without any progress on the stream. Reset to zero + * every time the stream returns a new resume token. + */ private int attempts; + + /** + * The value of {@link System#nanoTime()} at the first failure of the current sequence of + * consecutive failed attempts, or -1 if there is no active failure sequence. Reset to -1 every + * time the stream returns a new resume token. Used to enforce {@link + * RetrySettings#getTotalTimeout()} for non-default retry settings. + */ + private long retrySequenceStartNanos = -1L; + private ByteString resumeToken; private boolean finished; private final XGoogSpannerRequestId requestId; @@ -123,8 +154,45 @@ protected ResumableStreamIterator( this.requestId = xGoogRequestIdCreator.nextRequestId(0); } + private boolean hasDefaultStreamingRetrySettings() { + return Objects.equals(streamingRetrySettings, DEFAULT_STREAMING_RETRY_SETTINGS); + } + + /** + * Returns true if the number of consecutive failed attempts has reached the maximum number of + * attempts in the retry settings. {@link RetrySettings#getMaxAttempts()} equal to zero means that + * no maximum has been set, and that the number of attempts is unlimited. This is also the value + * in the default streaming retry settings, which means that only users who have explicitly opted + * in to a maximum number of attempts are affected by this limit. + */ + private boolean maxAttemptsExhausted() { + int maxAttempts = streamingRetrySettings.getMaxAttempts(); + return maxAttempts > 0 && attempts >= maxAttempts; + } + + /** + * Returns true if retrying after the proposed delay would exceed the total timeout in the retry + * settings. The total timeout limits the wall-clock time that is spent on a sequence of + * consecutive failed attempts without progress, measured from the first failure of the sequence. + * It is only enforced for non-default retry settings that set a positive total timeout: a total + * timeout of zero means that no total timeout has been set, and that only maxAttempts (if set) + * limits the retries. This mirrors the interpretation of these values in GAX. + */ + private boolean totalTimeoutExceeded(long proposedDelayMillis) { + if (hasDefaultStreamingRetrySettings()) { + return false; + } + long totalTimeoutMillis = streamingRetrySettings.getTotalTimeout().toMillis(); + if (totalTimeoutMillis <= 0L) { + return false; + } + long elapsedMillis = + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - retrySequenceStartNanos); + return elapsedMillis + Math.max(proposedDelayMillis, 0L) >= totalTimeoutMillis; + } + private ExponentialBackOff newBackOff() { - if (Objects.equals(streamingRetrySettings, DEFAULT_STREAMING_RETRY_SETTINGS)) { + if (hasDefaultStreamingRetrySettings()) { return new ExponentialBackOff.Builder() .setMultiplier(streamingRetrySettings.getRetryDelayMultiplier()) .setInitialIntervalMillis( @@ -150,19 +218,14 @@ private ExponentialBackOff newBackOff() { (int) Math.min( streamingRetrySettings.getMaxRetryDelay().toMillis(), Integer.MAX_VALUE))) - .setMaxElapsedTimeMillis( - Math.max( - 1, - (int) - Math.min( - streamingRetrySettings.getTotalTimeout().toMillis(), Integer.MAX_VALUE))) + // The total timeout is enforced explicitly in computeNext(), based on the elapsed time + // since the first failure of the current retry sequence. Prevent the backoff from + // returning BackOff.STOP, as that would misinterpret a total timeout of zero (that is, no + // total timeout) as a total timeout of one millisecond. + .setMaxElapsedTimeMillis(Integer.MAX_VALUE) .build(); } - private void backoffSleep(Context context, BackOff backoff) throws SpannerException { - backoffSleep(context, nextBackOffMillis(backoff)); - } - private static long nextBackOffMillis(BackOff backoff) throws SpannerException { try { return backoff.nextBackOffMillis(); @@ -263,6 +326,14 @@ protected PartialResultSet computeNext() { PartialResultSet next = stream.next(); boolean hasResumeToken = !next.getResumeToken().isEmpty(); if (hasResumeToken) { + // Only a resume token that differs from the last seen token is progress: a stream + // that repeatedly returns the token that was used to resume it has not moved past it. + if (!next.getResumeToken().equals(resumeToken)) { + // The stream made progress, so reset the budget for consecutive failed attempts. + attempts = 0; + backOff = null; + retrySequenceStartNanos = -1L; + } resumeToken = next.getResumeToken(); safeToRetry = true; } @@ -287,6 +358,36 @@ protected PartialResultSet computeNext() { } } catch (SpannerException spannerException) { if (safeToRetry && isRetryable(spannerException)) { + if (retrySequenceStartNanos == -1L) { + retrySequenceStartNanos = System.nanoTime(); + } + attempts++; + if (maxAttemptsExhausted()) { + span.addAnnotation( + "Stream broken. Not retrying because the maximum number of attempts has been" + + " exhausted", + spannerException); + span.setStatus(spannerException); + throw spannerException; + } + // Determine the retry delay: either the delay that the server included in the error, or + // otherwise a delay determined by the exponential backoff. + long delayMillis = spannerException.getRetryDelayInMillis(); + if (delayMillis == -1L) { + if (this.backOff == null) { + this.backOff = newBackOff(); + } + delayMillis = nextBackOffMillis(this.backOff); + } + // The total timeout budget applies regardless of whether the delay came from the server + // or from the backoff. + if (totalTimeoutExceeded(delayMillis)) { + span.addAnnotation( + "Stream broken. Not retrying because the total timeout has been exhausted", + spannerException); + span.setStatus(spannerException); + throw spannerException; + } span.addAnnotation("Stream broken. Safe to retry", spannerException); logger.log(Level.FINE, "Retryable exception, will sleep and retry", spannerException); // Truncate any items in the buffer before the last retry token. @@ -296,15 +397,7 @@ protected PartialResultSet computeNext() { assert buffer.isEmpty() || buffer.getLast().getResumeToken().equals(resumeToken); stream = null; try (IScope s = tracer.withSpan(span)) { - long delay = spannerException.getRetryDelayInMillis(); - if (delay != -1) { - backoffSleep(context, delay); - } else { - if (this.backOff == null) { - this.backOff = newBackOff(); - } - backoffSleep(context, this.backOff); - } + backoffSleep(context, delayMillis); } continue; diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java index cc44ba2f3f81..f725be0ce4f0 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java @@ -517,6 +517,19 @@ public static SimulatedExecutionTime ofStreamException(Exception exception, long 0, 0, Collections.singletonList(exception), false, Collections.singleton(streamIndex)); } + /** + * Creates a {@link SimulatedExecutionTime} that throws the given exceptions at the given + * indices in the returned stream. The exceptions and stream indices are matched by position: + * the first exception is thrown when the first call reaches the first stream index, the second + * exception when the next call reaches the second stream index, and so on. The stream index is + * reset for each (retried) call. + */ + public static SimulatedExecutionTime ofStreamExceptions( + Collection exceptions, Collection streamIndices) { + Preconditions.checkArgument(exceptions.size() == streamIndices.size()); + return new SimulatedExecutionTime(0, 0, exceptions, false, streamIndices); + } + public static SimulatedExecutionTime stickyDatabaseNotFoundException(String name) { return ofStickyException( SpannerExceptionFactoryTest.newStatusDatabaseNotFoundException(name)); diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java index f13c0bb1237a..81a05f051338 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java @@ -24,9 +24,11 @@ import static org.mockito.Mockito.when; import com.google.api.client.util.BackOff; +import com.google.api.gax.retrying.RetrySettings; import com.google.cloud.spanner.ErrorHandler.DefaultErrorHandler; import com.google.cloud.spanner.XGoogSpannerRequestId.NoopRequestIdCreator; import com.google.cloud.spanner.v1.stub.SpannerStubSettings; +import com.google.common.base.Stopwatch; import com.google.common.collect.AbstractIterator; import com.google.common.collect.ImmutableList; import com.google.common.collect.Lists; @@ -159,7 +161,12 @@ public void setUp() { } private void initWithLimit(int maxBufferSize) { + initWithLimitAndRetrySettings( + maxBufferSize, + SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetrySettings()); + } + private void initWithLimitAndRetrySettings(int maxBufferSize, RetrySettings retrySettings) { resumableStreamIterator = new ResumableStreamIterator( maxBufferSize, @@ -167,7 +174,7 @@ private void initWithLimit(int maxBufferSize) { new OpenTelemetrySpan(mock(io.opentelemetry.api.trace.Span.class)), new TraceWrapper(Tracing.getTracer(), OpenTelemetry.noop().getTracer(""), false), DefaultErrorHandler.INSTANCE, - SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetrySettings(), + retrySettings, SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetryableCodes(), NoopRequestIdCreator.INSTANCE) { @Override @@ -313,24 +320,199 @@ public void retryableErrorWithoutRetryInfo() throws IOException { setInternalState( ResumableStreamIterator.class, this.resumableStreamIterator, "backOff", backOff); + // The first stream fails before returning any resume token: receiving a new resume token + // resets the backoff by design, which would discard the injected mock backoff. ResultSetStream s1 = Mockito.mock(ResultSetStream.class); - Mockito.when(starter.startStream(null, null)).thenReturn(new ResultSetIterator(s1)); Mockito.when(s1.next()) - .thenReturn(resultSet(ByteString.copyFromUtf8("r1"), "a")) .thenThrow( new RetryableException( ErrorCode.UNAVAILABLE, "failed by test", Status.UNAVAILABLE.asRuntimeException())); ResultSetStream s2 = Mockito.mock(ResultSetStream.class); - Mockito.when(starter.startStream(ByteString.copyFromUtf8("r1"), null)) - .thenReturn(new ResultSetIterator(s2)); Mockito.when(s2.next()) + .thenReturn(resultSet(ByteString.copyFromUtf8("r1"), "a")) .thenReturn(resultSet(ByteString.copyFromUtf8("r2"), "b")) .thenReturn(null); + + Mockito.when(starter.startStream(null, null)) + .thenReturn(new ResultSetIterator(s1)) + .thenReturn(new ResultSetIterator(s2)); assertThat(consume(resumableStreamIterator)).containsExactly("a", "b").inOrder(); verify(backOff).nextBackOffMillis(); } + @Test(timeout = 60000L) + public void customMaxAttempts_stopsResumeAfterMaxAttempts() { + initWithLimitAndRetrySettings( + Integer.MAX_VALUE, + RetrySettings.newBuilder() + .setInitialRetryDelayDuration(java.time.Duration.ofMillis(1L)) + .setMaxRetryDelayDuration(java.time.Duration.ofMillis(1L)) + .setRetryDelayMultiplier(1.0) + .setMaxAttempts(2) + .setTotalTimeoutDuration(java.time.Duration.ofSeconds(30L)) + .build()); + + // Every stream fails with a retryable error before returning any data. Without a bound on the + // number of attempts, this loops forever. + Mockito.when(starter.startStream(Mockito.any(), Mockito.any())) + .thenAnswer( + invocation -> { + ResultSetStream stream = Mockito.mock(ResultSetStream.class); + Mockito.when(stream.next()) + .thenThrow(new RetryableException(errorCodeParameter, "failed by test")); + return new ResultSetIterator(stream); + }); + + SpannerException e = + assertThrows(SpannerException.class, () -> consume(resumableStreamIterator)); + assertEquals(errorCodeParameter, e.getErrorCode()); + Mockito.verify(starter, Mockito.times(2)).startStream(Mockito.any(), Mockito.any()); + } + + @Test(timeout = 60000L) + public void customMaxAttemptsWithoutTotalTimeout_makesExactlyMaxAttempts() { + // A total timeout of zero means that no total timeout has been set. The number of attempts + // must then be limited by maxAttempts alone: the unset total timeout must not be interpreted + // as a (near-)zero time budget that stops the retries before maxAttempts has been reached. + initWithLimitAndRetrySettings( + Integer.MAX_VALUE, + RetrySettings.newBuilder() + .setInitialRetryDelayDuration(java.time.Duration.ofMillis(5L)) + .setMaxRetryDelayDuration(java.time.Duration.ofMillis(5L)) + .setRetryDelayMultiplier(1.0) + .setMaxAttempts(3) + .build()); + + // Every stream fails with a retryable error without retry info, so the exponential backoff + // determines the retry delays. + Mockito.when(starter.startStream(Mockito.any(), Mockito.any())) + .thenAnswer( + invocation -> { + ResultSetStream stream = Mockito.mock(ResultSetStream.class); + Mockito.when(stream.next()) + .thenThrow( + new RetryableException( + errorCodeParameter, + "failed by test", + errorCodeParameter.getGrpcStatus().asRuntimeException())); + return new ResultSetIterator(stream); + }); + + SpannerException e = + assertThrows(SpannerException.class, () -> consume(resumableStreamIterator)); + assertEquals(errorCodeParameter, e.getErrorCode()); + Mockito.verify(starter, Mockito.times(3)).startStream(Mockito.any(), Mockito.any()); + } + + @Test(timeout = 60000L) + public void repeatedIdenticalResumeToken_doesNotResetAttempts() { + // A stream that repeatedly returns the same resume token has not made any progress: only a + // new resume token resets the budget for consecutive failed attempts. Without this, a stream + // that always returns the token that was used to resume it and then fails would retry + // forever, regardless of maxAttempts. + initWithLimitAndRetrySettings( + Integer.MAX_VALUE, + RetrySettings.newBuilder() + .setInitialRetryDelayDuration(java.time.Duration.ofMillis(1L)) + .setMaxRetryDelayDuration(java.time.Duration.ofMillis(1L)) + .setRetryDelayMultiplier(1.0) + .setMaxAttempts(2) + .setTotalTimeoutDuration(java.time.Duration.ofSeconds(30L)) + .build()); + + ByteString token = ByteString.copyFromUtf8("r1"); + ResultSetStream s1 = Mockito.mock(ResultSetStream.class); + Mockito.when(s1.next()) + .thenReturn(resultSet(token, "a")) + .thenThrow(new RetryableException(errorCodeParameter, "failed by test")); + Mockito.when(starter.startStream(null, null)).thenReturn(new ResultSetIterator(s1)); + // Every resumed stream returns the same resume token again and then fails. + Mockito.when(starter.startStream(token, null)) + .thenAnswer( + invocation -> { + ResultSetStream stream = Mockito.mock(ResultSetStream.class); + Mockito.when(stream.next()) + .thenReturn(resultSet(token, "x")) + .thenThrow(new RetryableException(errorCodeParameter, "failed by test")); + return new ResultSetIterator(stream); + }); + + SpannerException e = + assertThrows(SpannerException.class, () -> consume(resumableStreamIterator)); + assertEquals(errorCodeParameter, e.getErrorCode()); + Mockito.verify(starter, Mockito.times(1)).startStream(null, null); + Mockito.verify(starter, Mockito.times(1)).startStream(token, null); + } + + @Test(timeout = 60000L) + public void customTotalTimeoutSmallerThanRetryDelay_doesNotRetry() { + // Retrying is only allowed if the retry delay still fits in the remaining total timeout + // budget. A retry delay that is larger than the total timeout means that the first failure + // already exhausts the budget. + initWithLimitAndRetrySettings( + Integer.MAX_VALUE, + RetrySettings.newBuilder() + .setInitialRetryDelayDuration(java.time.Duration.ofSeconds(1L)) + .setMaxRetryDelayDuration(java.time.Duration.ofSeconds(1L)) + .setRetryDelayMultiplier(1.0) + .setTotalTimeoutDuration(java.time.Duration.ofMillis(100L)) + .build()); + + Mockito.when(starter.startStream(Mockito.any(), Mockito.any())) + .thenAnswer( + invocation -> { + ResultSetStream stream = Mockito.mock(ResultSetStream.class); + Mockito.when(stream.next()) + .thenThrow( + new RetryableException( + errorCodeParameter, + "failed by test", + errorCodeParameter.getGrpcStatus().asRuntimeException())); + return new ResultSetIterator(stream); + }); + + Stopwatch stopwatch = Stopwatch.createStarted(); + SpannerException e = + assertThrows(SpannerException.class, () -> consume(resumableStreamIterator)); + assertEquals(errorCodeParameter, e.getErrorCode()); + Mockito.verify(starter, Mockito.times(1)).startStream(Mockito.any(), Mockito.any()); + // The one-second retry delay must not have been slept before giving up. + assertThat(stopwatch.elapsed(TimeUnit.MILLISECONDS)).isLessThan(5000L); + } + + @Test(timeout = 60000L) + public void customTotalTimeout_stopsResumeWhenTotalTimeoutIsExhausted() { + initWithLimitAndRetrySettings( + Integer.MAX_VALUE, + RetrySettings.newBuilder() + .setInitialRetryDelayDuration(java.time.Duration.ofMillis(1L)) + .setMaxRetryDelayDuration(java.time.Duration.ofMillis(10L)) + .setRetryDelayMultiplier(1.0) + .setTotalTimeoutDuration(java.time.Duration.ofMillis(50L)) + .build()); + + // Every stream fails with a retryable error without retry info, so the exponential backoff + // determines the retry delays. Without a bound on the total time spent retrying, this loops + // forever. + Mockito.when(starter.startStream(Mockito.any(), Mockito.any())) + .thenAnswer( + invocation -> { + ResultSetStream stream = Mockito.mock(ResultSetStream.class); + Mockito.when(stream.next()) + .thenThrow( + new RetryableException( + errorCodeParameter, + "failed by test", + errorCodeParameter.getGrpcStatus().asRuntimeException())); + return new ResultSetIterator(stream); + }); + + SpannerException e = + assertThrows(SpannerException.class, () -> consume(resumableStreamIterator)); + assertEquals(errorCodeParameter, e.getErrorCode()); + } + @Test public void nonRetryableError() { ResultSetStream s1 = Mockito.mock(ResultSetStream.class); diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java new file mode 100644 index 000000000000..5c62f7cc10f2 --- /dev/null +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java @@ -0,0 +1,357 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.spanner; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import com.google.api.gax.grpc.testing.LocalChannelProvider; +import com.google.api.gax.retrying.RetrySettings; +import com.google.api.gax.rpc.StatusCode.Code; +import com.google.cloud.NoCredentials; +import com.google.cloud.spanner.MockSpannerServiceImpl.SimulatedExecutionTime; +import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult; +import com.google.common.base.Stopwatch; +import com.google.protobuf.ListValue; +import com.google.rpc.RetryInfo; +import com.google.spanner.v1.ExecuteSqlRequest; +import com.google.spanner.v1.ResultSetMetadata; +import com.google.spanner.v1.StructType; +import com.google.spanner.v1.StructType.Field; +import com.google.spanner.v1.TypeCode; +import io.grpc.Metadata; +import io.grpc.Server; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.protobuf.ProtoUtils; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** + * Tests that the streaming resume loop in {@link ResumableStreamIterator} honors the {@link + * RetrySettings#getMaxAttempts()} and {@link RetrySettings#getTotalTimeout()} that have been + * configured for ExecuteStreamingSql, instead of retrying indefinitely, while the default settings + * (no maximum number of attempts) keep the existing unbounded resume behavior. + * + *

See https://github.com/googleapis/google-cloud-java/issues/12255. The tests in this class + * drive a real {@link Spanner} client against an in-process mock Spanner server, so the entire + * client stack (client -> GAX -> resume loop) is exercised. + */ +@RunWith(JUnit4.class) +public class StreamingRetryBudgetMockServerTest { + private static final Statement SELECT_QUERY = Statement.of("SELECT C FROM T"); + private static final int ROW_COUNT = 4; + private static final ResultSetMetadata METADATA = + ResultSetMetadata.newBuilder() + .setRowType( + StructType.newBuilder() + .addFields( + Field.newBuilder() + .setName("C") + .setType( + com.google.spanner.v1.Type.newBuilder() + .setCode(TypeCode.INT64) + .build()) + .build()) + .build()) + .build(); + private static final StatusRuntimeException UNAVAILABLE = + Status.UNAVAILABLE.withDescription("Retryable test exception.").asRuntimeException(); + private static final StatusRuntimeException DEADLINE_EXCEEDED = + Status.DEADLINE_EXCEEDED.withDescription("Test deadline exceeded.").asRuntimeException(); + private static final StatusRuntimeException RESOURCE_EXHAUSTED_WITH_RETRY_DELAY = + Status.RESOURCE_EXHAUSTED + .withDescription("Retryable test exception with retry delay.") + .asRuntimeException(createRetryInfoTrailers()); + + private static Metadata createRetryInfoTrailers() { + Metadata trailers = new Metadata(); + RetryInfo retryInfo = + RetryInfo.newBuilder() + .setRetryDelay( + com.google.protobuf.Duration.newBuilder() + .setNanos((int) TimeUnit.MILLISECONDS.toNanos(10L)) + .setSeconds(0L) + .build()) + .build(); + trailers.put(ProtoUtils.keyForProto(RetryInfo.getDefaultInstance()), retryInfo); + return trailers; + } + + private static MockSpannerServiceImpl mockSpanner; + private static Server server; + private static LocalChannelProvider channelProvider; + + private Spanner spanner; + private DatabaseClient client; + private Spanner spannerWithCustomRetrySettings; + private DatabaseClient clientWithCustomRetrySettings; + private Spanner spannerWithTotalTimeout; + private DatabaseClient clientWithTotalTimeout; + + private static com.google.spanner.v1.ResultSet createResultSet(int rowCount) { + com.google.spanner.v1.ResultSet.Builder builder = + com.google.spanner.v1.ResultSet.newBuilder().setMetadata(METADATA); + for (int row = 0; row < rowCount; row++) { + builder.addRows( + ListValue.newBuilder() + .addValues( + com.google.protobuf.Value.newBuilder() + .setStringValue(String.valueOf(row)) + .build()) + .build()); + } + return builder.build(); + } + + @BeforeClass + public static void startStaticServer() throws Exception { + mockSpanner = new MockSpannerServiceImpl(); + mockSpanner.setAbortProbability(0.0D); // We don't want any unpredictable aborted transactions. + mockSpanner.putStatementResult(StatementResult.query(SELECT_QUERY, createResultSet(ROW_COUNT))); + + String uniqueName = InProcessServerBuilder.generateName(); + server = + InProcessServerBuilder.forName(uniqueName) + // We need to use a real executor for timeouts to occur. + .scheduledExecutorService(new ScheduledThreadPoolExecutor(1)) + .addService(mockSpanner) + .build() + .start(); + channelProvider = LocalChannelProvider.create(uniqueName); + } + + @AfterClass + public static void stopServer() throws InterruptedException { + server.shutdown(); + server.awaitTermination(); + } + + @Before + public void setUp() { + mockSpanner.reset(); + mockSpanner.removeAllExecutionTimes(); + SpannerOptions.Builder builder = + SpannerOptions.newBuilder() + .setProjectId("[PROJECT]") + .setChannelProvider(channelProvider) + .setCredentials(NoCredentials.getInstance()); + // Make sure the session pool is empty by default. + SessionPoolOptions sessionPoolOptions = + SessionPoolOptions.newBuilder().setMinSessions(0).build(); + // Add a wait time for sessions to be initialized. In this case, since minSessions = 0, the + // wait time is for multiplexed sessions. + if (sessionPoolOptions.getUseMultiplexedSession()) { + sessionPoolOptions = + sessionPoolOptions.toBuilder() + .setWaitForMinSessionsDuration(Duration.ofSeconds(5)) + .build(); + } + builder.setSessionPoolOption(sessionPoolOptions); + + // A client with the default retry settings for ExecuteStreamingSql. These settings do not set + // a maximum number of attempts, meaning that the number of resume attempts is unlimited. + spanner = builder.build().getService(); + client = spanner.getDatabaseClient(DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE]")); + + // A client with custom resiliency settings for ExecuteStreamingSql, similar to what a user + // would configure: a small number of attempts, a per-attempt timeout, and DEADLINE_EXCEEDED + // (and others) as retryable codes. The total timeout is deliberately kept large, so the tests + // verify that maxAttempts is the limit that stops the retries. + RetrySettings customRetrySettings = + RetrySettings.newBuilder() + .setInitialRetryDelayDuration(Duration.ofMillis(1L)) + .setMaxRetryDelayDuration(Duration.ofMillis(1L)) + .setRetryDelayMultiplier(1.0) + .setInitialRpcTimeoutDuration(Duration.ofMillis(500L)) + .setMaxRpcTimeoutDuration(Duration.ofMillis(500L)) + .setMaxAttempts(2) + .setTotalTimeoutDuration(Duration.ofSeconds(30L)) + .build(); + builder + .getSpannerStubSettingsBuilder() + .executeStreamingSqlSettings() + .setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED) + .setRetrySettings(customRetrySettings); + spannerWithCustomRetrySettings = builder.build().getService(); + clientWithCustomRetrySettings = + spannerWithCustomRetrySettings.getDatabaseClient( + DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE]")); + + // A client that does not set a maximum number of attempts, but that does set a small total + // timeout. The total timeout is the limit that stops the retries for this client. + RetrySettings totalTimeoutRetrySettings = + RetrySettings.newBuilder() + .setInitialRetryDelayDuration(Duration.ofMillis(20L)) + .setMaxRetryDelayDuration(Duration.ofMillis(20L)) + .setRetryDelayMultiplier(1.0) + .setInitialRpcTimeoutDuration(Duration.ofMillis(500L)) + .setMaxRpcTimeoutDuration(Duration.ofMillis(500L)) + .setTotalTimeoutDuration(Duration.ofMillis(200L)) + .build(); + builder + .getSpannerStubSettingsBuilder() + .executeStreamingSqlSettings() + .setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED) + .setRetrySettings(totalTimeoutRetrySettings); + spannerWithTotalTimeout = builder.build().getService(); + clientWithTotalTimeout = + spannerWithTotalTimeout.getDatabaseClient( + DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE]")); + } + + @After + public void tearDown() { + spannerWithTotalTimeout.close(); + spannerWithCustomRetrySettings.close(); + spanner.close(); + } + + /** + * A query where every ExecuteStreamingSql attempt fails with DEADLINE_EXCEEDED must fail after + * the configured maximum number of attempts. Without a bounded resume loop in {@link + * ResumableStreamIterator} this test never finishes: every failed stream is resumed with a new + * RPC, regardless of the configured maxAttempts and totalTimeout. The timeout on this test + * ensures that such a regression fails the test instead of hanging the build. + */ + @Test(timeout = 60000L) + public void maxAttemptsExhausted_stopsStreamingRetries() { + mockSpanner.setExecuteStreamingSqlExecutionTime( + SimulatedExecutionTime.ofStickyException(DEADLINE_EXCEEDED)); + mockSpanner.clearRequests(); + + try (ResultSet resultSet = + clientWithCustomRetrySettings.singleUse().executeQuery(SELECT_QUERY)) { + SpannerException exception = assertThrows(SpannerException.class, resultSet::next); + assertEquals(ErrorCode.DEADLINE_EXCEEDED, exception.getErrorCode()); + } + // The retry settings are applied independently by two layers: the resume loop in + // ResumableStreamIterator starts at most maxAttempts (2) streams, and each of those streams + // is a GAX call that itself retries attempts that fail before the first response up to + // maxAttempts (2) times. A configured maxAttempts of N therefore bounds the number of RPC + // attempts without progress by N * N (here: 4), not by N. See the class documentation of + // ResumableStreamIterator. + assertEquals(4, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class)); + } + + /** + * A query where every ExecuteStreamingSql attempt fails with a retryable error must fail once + * the configured total timeout has been exhausted, also if no maximum number of attempts has + * been set. Without a bounded resume loop this test never finishes. + */ + @Test(timeout = 60000L) + public void totalTimeoutExhausted_stopsStreamingRetries() { + mockSpanner.setExecuteStreamingSqlExecutionTime( + SimulatedExecutionTime.ofStickyException(UNAVAILABLE)); + mockSpanner.clearRequests(); + + Stopwatch stopwatch = Stopwatch.createStarted(); + try (ResultSet resultSet = clientWithTotalTimeout.singleUse().executeQuery(SELECT_QUERY)) { + SpannerException exception = assertThrows(SpannerException.class, resultSet::next); + assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode()); + } + // The exact number of attempts depends on timing, but the query must have been retried at + // least once, and must have failed shortly after the total timeout (200ms) elapsed. The + // wall-clock bound is kept well above the total timeout to avoid flakiness on slow CI + // machines, but far below the unbounded behavior that the fix prevents. + assertTrue(mockSpanner.countRequestsOfType(ExecuteSqlRequest.class) > 1); + assertTrue(stopwatch.elapsed(TimeUnit.MILLISECONDS) < 10000L); + } + + /** + * The total timeout must also be enforced when the retryable errors carry a server-supplied + * retry delay (RetryInfo), as such errors bypass the exponential backoff. Without that, a + * client with no maximum number of attempts and a total timeout would retry forever when the + * server keeps returning, for example, RESOURCE_EXHAUSTED with a retry delay. Without a bounded + * resume loop this test never finishes. + */ + @Test(timeout = 60000L) + public void totalTimeoutExhausted_withServerSuppliedRetryDelay_stopsStreamingRetries() { + mockSpanner.setExecuteStreamingSqlExecutionTime( + SimulatedExecutionTime.ofStickyException(RESOURCE_EXHAUSTED_WITH_RETRY_DELAY)); + mockSpanner.clearRequests(); + + Stopwatch stopwatch = Stopwatch.createStarted(); + try (ResultSet resultSet = clientWithTotalTimeout.singleUse().executeQuery(SELECT_QUERY)) { + SpannerException exception = assertThrows(SpannerException.class, resultSet::next); + assertEquals(ErrorCode.RESOURCE_EXHAUSTED, exception.getErrorCode()); + } + assertTrue(mockSpanner.countRequestsOfType(ExecuteSqlRequest.class) > 1); + assertTrue(stopwatch.elapsed(TimeUnit.MILLISECONDS) < 10000L); + } + + /** + * The default retry settings do not set a maximum number of attempts. A client that uses the + * default settings must keep today's behavior: the stream is resumed as often as needed, also + * when the number of consecutive failures is higher than any small maximum. + */ + @Test(timeout = 60000L) + public void defaultRetrySettings_keepsUnboundedResumes() { + mockSpanner.setExecuteStreamingSqlExecutionTime( + SimulatedExecutionTime.ofExceptions(Collections.nCopies(5, UNAVAILABLE))); + mockSpanner.clearRequests(); + + int rows = 0; + try (ResultSet resultSet = client.singleUse().executeQuery(SELECT_QUERY)) { + while (resultSet.next()) { + rows++; + } + } + assertEquals(ROW_COUNT, rows); + // 5 failed attempts + 1 successful attempt. + assertEquals(6, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class)); + } + + /** + * Only consecutive failures count against the maximum number of attempts: any progress on the + * stream resets the budget. A stream that fails, resumes and makes progress, and then fails + * again must succeed with maxAttempts=2, as the two failures are not consecutive. + */ + @Test(timeout = 60000L) + public void progressOnStream_resetsAttempts() { + // Break the stream after the second PartialResultSet of the first call, and again after the + // second PartialResultSet of the second (resumed) call. Every PartialResultSet returned by the + // mock server contains a resume token, so the client sees progress between the two failures. + mockSpanner.setExecuteStreamingSqlExecutionTime( + SimulatedExecutionTime.ofStreamExceptions( + Arrays.asList(UNAVAILABLE, UNAVAILABLE), Arrays.asList(1L, 1L))); + mockSpanner.clearRequests(); + + int rows = 0; + try (ResultSet resultSet = + clientWithCustomRetrySettings.singleUse().executeQuery(SELECT_QUERY)) { + while (resultSet.next()) { + rows++; + } + } + assertEquals(ROW_COUNT, rows); + // The initial attempt and two resumed attempts. + assertEquals(3, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class)); + } +} From d2bd4b196ca3332a8df849f5dace7795bc3d401b Mon Sep 17 00:00:00 2001 From: Rahul Yadav Date: Mon, 14 Sep 2026 13:36:22 +0530 Subject: [PATCH 2/4] fix(spanner): address review feedback on streaming retry budget --- .../spanner/ResumableStreamIterator.java | 15 +++-- .../spanner/ResumableStreamIteratorTest.java | 56 +++++++++++++++++++ .../StreamingRetryBudgetMockServerTest.java | 16 +++++- 3 files changed, 79 insertions(+), 8 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java index 88a01114e0a3..a2b36aa17b6f 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java @@ -95,9 +95,8 @@ abstract class ResumableStreamIterator extends AbstractIterator= totalTimeoutMillis; + if (elapsedMillis < 0L) { + elapsedMillis = 0L; + } + if (elapsedMillis >= totalTimeoutMillis) { + return true; + } + return Math.max(proposedDelayMillis, 0L) >= totalTimeoutMillis - elapsedMillis; } private ExponentialBackOff newBackOff() { @@ -358,7 +363,7 @@ protected PartialResultSet computeNext() { } } catch (SpannerException spannerException) { if (safeToRetry && isRetryable(spannerException)) { - if (retrySequenceStartNanos == -1L) { + if (attempts == 0) { retrySequenceStartNanos = System.nanoTime(); } attempts++; diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java index 81a05f051338..c1edf6a056cb 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java @@ -47,6 +47,7 @@ import io.opentelemetry.context.Scope; import java.io.IOException; import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Iterator; import java.util.LinkedList; @@ -445,6 +446,61 @@ public void repeatedIdenticalResumeToken_doesNotResetAttempts() { Mockito.verify(starter, Mockito.times(1)).startStream(token, null); } + @Test + public void activeRetrySequence_preservesNegativeOneStartTime() throws Exception { + initWithLimitAndRetrySettings( + Integer.MAX_VALUE, RetrySettings.newBuilder().setMaxAttempts(2).build()); + Field attempts = ResumableStreamIterator.class.getDeclaredField("attempts"); + attempts.setAccessible(true); + attempts.setInt(resumableStreamIterator, 1); + Field startNanos = ResumableStreamIterator.class.getDeclaredField("retrySequenceStartNanos"); + startNanos.setAccessible(true); + startNanos.setLong(resumableStreamIterator, -1L); + ResultSetStream stream = Mockito.mock(ResultSetStream.class); + Mockito.when(stream.next()) + .thenThrow(new RetryableException(errorCodeParameter, "failed by test")); + Mockito.when(starter.startStream(null, null)).thenReturn(new ResultSetIterator(stream)); + + assertThrows(SpannerException.class, () -> consume(resumableStreamIterator)); + + assertEquals(-1L, startNanos.getLong(resumableStreamIterator)); + } + + @Test + public void totalTimeout_largeDelayDoesNotOverflow() throws Exception { + assertThat(totalTimeoutExceeded(1000L, -100L, Long.MAX_VALUE)).isTrue(); + } + + @Test + public void totalTimeout_negativeElapsedIsClamped() throws Exception { + assertThat(totalTimeoutExceeded(Long.MAX_VALUE, 60000L, Long.MAX_VALUE)).isTrue(); + assertThat(totalTimeoutExceeded(Long.MAX_VALUE, 60000L, 0L)).isFalse(); + } + + @Test + public void totalTimeout_elapsedBudgetAndNegativeDelay() throws Exception { + assertThat(totalTimeoutExceeded(1000L, -60000L, -1L)).isTrue(); + assertThat(totalTimeoutExceeded(1000L, 60000L, -1L)).isFalse(); + assertThat(totalTimeoutExceeded(1000L, 60000L, 1000L)).isTrue(); + } + + private boolean totalTimeoutExceeded( + long timeoutMillis, long startOffsetMillis, long delayMillis) throws Exception { + initWithLimitAndRetrySettings( + Integer.MAX_VALUE, + RetrySettings.newBuilder() + .setTotalTimeoutDuration(java.time.Duration.ofMillis(timeoutMillis)) + .build()); + Field startNanos = ResumableStreamIterator.class.getDeclaredField("retrySequenceStartNanos"); + startNanos.setAccessible(true); + startNanos.setLong( + resumableStreamIterator, System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(startOffsetMillis)); + Method totalTimeoutExceeded = + ResumableStreamIterator.class.getDeclaredMethod("totalTimeoutExceeded", long.class); + totalTimeoutExceeded.setAccessible(true); + return (boolean) totalTimeoutExceeded.invoke(resumableStreamIterator, delayMillis); + } + @Test(timeout = 60000L) public void customTotalTimeoutSmallerThanRetryDelay_doesNotRetry() { // Retrying is only allowed if the retry delay still fits in the remaining total timeout diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java index 5c62f7cc10f2..322162f690e7 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java @@ -106,6 +106,7 @@ private static Metadata createRetryInfoTrailers() { private static MockSpannerServiceImpl mockSpanner; private static Server server; + private static ScheduledThreadPoolExecutor scheduledExecutor; private static LocalChannelProvider channelProvider; private Spanner spanner; @@ -136,11 +137,12 @@ public static void startStaticServer() throws Exception { mockSpanner.setAbortProbability(0.0D); // We don't want any unpredictable aborted transactions. mockSpanner.putStatementResult(StatementResult.query(SELECT_QUERY, createResultSet(ROW_COUNT))); + scheduledExecutor = new ScheduledThreadPoolExecutor(1); String uniqueName = InProcessServerBuilder.generateName(); server = InProcessServerBuilder.forName(uniqueName) // We need to use a real executor for timeouts to occur. - .scheduledExecutorService(new ScheduledThreadPoolExecutor(1)) + .scheduledExecutorService(scheduledExecutor) .addService(mockSpanner) .build() .start(); @@ -149,8 +151,16 @@ public static void startStaticServer() throws Exception { @AfterClass public static void stopServer() throws InterruptedException { - server.shutdown(); - server.awaitTermination(); + if (server != null) { + server.shutdown(); + server.awaitTermination(); + } + if (scheduledExecutor != null) { + scheduledExecutor.shutdown(); + if (!scheduledExecutor.awaitTermination(10, TimeUnit.SECONDS)) { + scheduledExecutor.shutdownNow(); + } + } } @Before From 4487994f2ac81b70282e65edc143ec31661a8404 Mon Sep 17 00:00:00 2001 From: Rahul Yadav Date: Mon, 14 Sep 2026 13:58:43 +0530 Subject: [PATCH 3/4] fix lint --- .../spanner/ResumableStreamIterator.java | 25 +++++++++---------- .../spanner/ResumableStreamIteratorTest.java | 7 +++--- .../StreamingRetryBudgetMockServerTest.java | 20 +++++++-------- 3 files changed, 26 insertions(+), 26 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java index a2b36aa17b6f..7f717933bd4f 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java @@ -57,18 +57,18 @@ * settings: consecutive failed attempts are limited by {@link RetrySettings#getMaxAttempts()} if * that has been set to a value greater than zero, and the total time spent on a sequence of * consecutive failed attempts is limited by {@link RetrySettings#getTotalTimeout()} if that has - * been set to a positive value. Only consecutive failures count against these budgets: any - * progress on the stream (that is, receiving a resume token that differs from the last seen resume - * token) resets both, so a long-running stream that regularly makes progress is not terminated by - * an occasional transient error. + * been set to a positive value. Only consecutive failures count against these budgets: any progress + * on the stream (that is, receiving a resume token that differs from the last seen resume token) + * resets both, so a long-running stream that regularly makes progress is not terminated by an + * occasional transient error. * - *

These limits bound the number of streams that this iterator starts. Each (re)started stream - * is a call through the underlying GAX callable, and GAX applies the same {@link RetrySettings} to + *

These limits bound the number of streams that this iterator starts. Each (re)started stream is + * a call through the underlying GAX callable, and GAX applies the same {@link RetrySettings} to * attempts of that call that fail before any response has been received. A stream that repeatedly * fails before its first response can therefore consist of up to maxAttempts RPC attempts itself, - * so a configured maxAttempts of N bounds the total number of RPC attempts without progress by - * N*N, not by N. The total timeout is measured in wall-clock time from the first failure of the - * sequence and therefore spans both layers. + * so a configured maxAttempts of N bounds the total number of RPC attempts without progress by N*N, + * not by N. The total timeout is measured in wall-clock time from the first failure of the sequence + * and therefore spans both layers. */ @VisibleForTesting abstract class ResumableStreamIterator extends AbstractIterator @@ -95,8 +95,8 @@ abstract class ResumableStreamIterator extends AbstractIterator Date: Mon, 14 Sep 2026 21:28:45 +0530 Subject: [PATCH 4/4] fix(spanner): address reviewer feedback on streaming retry budget - Normalize generated streaming retry defaults at the RPC boundary. - Extract retry handling and replace reflection with deterministic tests. - Document and test maxAttempts=1 as the way to disable streaming retries. - Cover StreamingRead attempt and timeout budgets with mock-server tests. --- .../spanner/ResumableStreamIterator.java | 146 +++++++------- .../google/cloud/spanner/SpannerOptions.java | 9 + .../cloud/spanner/spi/v1/GapicSpannerRpc.java | 24 ++- .../spanner/ResumableStreamIteratorTest.java | 185 ++++++++++-------- .../StreamingRetryBudgetMockServerTest.java | 132 ++++++++++++- .../spanner/spi/v1/GapicSpannerRpcTest.java | 78 ++++++++ 6 files changed, 412 insertions(+), 162 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java index 7f717933bd4f..d6477fe65104 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ResumableStreamIterator.java @@ -28,7 +28,6 @@ import com.google.api.gax.retrying.RetrySettings; import com.google.api.gax.rpc.StatusCode.Code; import com.google.cloud.spanner.AbstractResultSet.CloseableIterator; -import com.google.cloud.spanner.v1.stub.SpannerStubSettings; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.AbstractIterator; @@ -38,7 +37,6 @@ import io.opentelemetry.api.common.Attributes; import java.io.IOException; import java.util.LinkedList; -import java.util.Objects; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; @@ -53,14 +51,15 @@ * have a resume token until one is seen or buffer space is exceeded, which reduces the chance of * yielding data to the caller that cannot be resumed. * - *

When non-default streaming retry settings are used, the resume loop is bounded by those - * settings: consecutive failed attempts are limited by {@link RetrySettings#getMaxAttempts()} if - * that has been set to a value greater than zero, and the total time spent on a sequence of - * consecutive failed attempts is limited by {@link RetrySettings#getTotalTimeout()} if that has - * been set to a positive value. Only consecutive failures count against these budgets: any progress - * on the stream (that is, receiving a resume token that differs from the last seen resume token) - * resets both, so a long-running stream that regularly makes progress is not terminated by an - * occasional transient error. + *

The resume loop is bounded by the streaming retry settings: consecutive failed attempts are + * limited by {@link RetrySettings#getMaxAttempts()} if that has been set to a value greater than + * zero, and the total time spent on a sequence of consecutive failed attempts is limited by {@link + * RetrySettings#getTotalTimeout()} if that has been set to a positive value. Only consecutive + * failures count against these budgets: any progress on the stream (that is, receiving a resume + * token that differs from the last seen resume token) resets both, so a long-running stream that + * regularly makes progress is not terminated by an occasional transient error. The default settings + * have no maximum number of attempts or total timeout. Setting maxAttempts to 1 disables streaming + * retries. * *

These limits bound the number of streams that this iterator starts. Each (re)started stream is * a call through the underlying GAX callable, and GAX applies the same {@link RetrySettings} to @@ -73,8 +72,6 @@ @VisibleForTesting abstract class ResumableStreamIterator extends AbstractIterator implements CloseableIterator { - private static final RetrySettings DEFAULT_STREAMING_RETRY_SETTINGS = - SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetrySettings(); private final ErrorHandler errorHandler; private AsyncResultSet.StreamMessageListener streamMessageListener; private final RetrySettings streamingRetrySettings; @@ -96,7 +93,7 @@ abstract class ResumableStreamIterator extends AbstractIterator 0. @@ -362,47 +342,7 @@ protected PartialResultSet computeNext() { } } catch (SpannerException spannerException) { if (safeToRetry && isRetryable(spannerException)) { - if (attempts == 0) { - retrySequenceStartNanos = System.nanoTime(); - } - attempts++; - if (maxAttemptsExhausted()) { - span.addAnnotation( - "Stream broken. Not retrying because the maximum number of attempts has been" - + " exhausted", - spannerException); - span.setStatus(spannerException); - throw spannerException; - } - // Determine the retry delay: either the delay that the server included in the error, or - // otherwise a delay determined by the exponential backoff. - long delayMillis = spannerException.getRetryDelayInMillis(); - if (delayMillis == -1L) { - if (this.backOff == null) { - this.backOff = newBackOff(); - } - delayMillis = nextBackOffMillis(this.backOff); - } - // The total timeout budget applies regardless of whether the delay came from the server - // or from the backoff. - if (totalTimeoutExceeded(delayMillis)) { - span.addAnnotation( - "Stream broken. Not retrying because the total timeout has been exhausted", - spannerException); - span.setStatus(spannerException); - throw spannerException; - } - span.addAnnotation("Stream broken. Safe to retry", spannerException); - logger.log(Level.FINE, "Retryable exception, will sleep and retry", spannerException); - // Truncate any items in the buffer before the last retry token. - while (!buffer.isEmpty() && buffer.getLast().getResumeToken().isEmpty()) { - buffer.removeLast(); - } - assert buffer.isEmpty() || buffer.getLast().getResumeToken().equals(resumeToken); - stream = null; - try (IScope s = tracer.withSpan(span)) { - backoffSleep(context, delayMillis); - } + handleRetryableException(context, spannerException); continue; } @@ -428,6 +368,62 @@ && prepareIteratorForRetryOnDifferentGrpcChannel()) { } } + /** Monotonic time source, overridable for deterministic retry-budget tests. */ + @VisibleForTesting + long nanoTime() { + return System.nanoTime(); + } + + @VisibleForTesting + long checkRetryBudgetAndGetDelay(SpannerException spannerException) { + if (attempts == 0) { + retrySequenceStartNanos = nanoTime(); + } + attempts++; + if (maxAttemptsExhausted()) { + span.addAnnotation( + "Stream broken. Not retrying because the maximum number of attempts has been" + + " exhausted", + spannerException); + span.setStatus(spannerException); + throw spannerException; + } + // Determine the retry delay: either the delay that the server included in the error, or + // otherwise a delay determined by the exponential backoff. + long delayMillis = spannerException.getRetryDelayInMillis(); + if (delayMillis == -1L) { + if (this.backOff == null) { + this.backOff = newBackOff(); + } + delayMillis = nextBackOffMillis(this.backOff); + } + // The total timeout budget applies regardless of whether the delay came from the server + // or from the backoff. + if (totalTimeoutExceeded(delayMillis)) { + span.addAnnotation( + "Stream broken. Not retrying because the total timeout has been exhausted", + spannerException); + span.setStatus(spannerException); + throw spannerException; + } + return delayMillis; + } + + private void handleRetryableException(Context context, SpannerException spannerException) { + long delayMillis = checkRetryBudgetAndGetDelay(spannerException); + span.addAnnotation("Stream broken. Safe to retry", spannerException); + logger.log(Level.FINE, "Retryable exception, will sleep and retry", spannerException); + // Truncate any items in the buffer before the last retry token. + while (!buffer.isEmpty() && buffer.getLast().getResumeToken().isEmpty()) { + buffer.removeLast(); + } + assert buffer.isEmpty() || buffer.getLast().getResumeToken().equals(resumeToken); + stream = null; + try (IScope s = tracer.withSpan(span)) { + backoffSleep(context, delayMillis); + } + } + private void startGrpcStreaming() { if (stream == null) { span.addAnnotation( diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java index a7d77a7006f4..ca1f9abec4f0 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java @@ -1669,6 +1669,15 @@ public Builder setRetrySettings(RetrySettings retrySettings) { * href="https://github.com/googleapis/googleapis/blob/master/google/spanner/v1/spanner_gapic.yaml">spanner_gapic.yaml. * Retries are configured for idempotent methods but not for non-idempotent methods. * + *

For streaming queries and reads, configure {@code executeStreamingSqlSettings()} and + * {@code streamingReadSettings()}, respectively. Set {@code maxAttempts=1} to disable streaming + * retries; an empty set of retryable codes does not disable retries for intrinsically retryable + * errors. Defaults allow unlimited streaming resumes. When customizing retry settings, set the + * total timeout explicitly: calling {@code toBuilder()} on the stub's retry settings copies + * GAPIC's generated one-hour {@code totalTimeout}; set it to zero explicitly for unlimited + * resumes. Limits apply to consecutive failures and reset when the stream makes progress by + * returning a new resume token. + * *

You can set the same {@link RetrySettings} for all unary methods by calling this: * * 


diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
index e76091148705..fcf7baec3608 100644
--- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
+++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
@@ -271,6 +271,15 @@ public class GapicSpannerRpc implements SpannerRpc {
   private static final CallOptions.Key BASE_CONTEXT_MARKER_KEY =
       CallOptions.Key.create("BASE_CONTEXT_MARKER_KEY");
 
+  // Normalize the generated placeholder to the historical streaming resume policy.
+  static final RetrySettings DEFAULT_STREAMING_RETRY_SETTINGS =
+      RetrySettings.newBuilder()
+          .setTotalTimeoutDuration(Duration.ZERO)
+          .setMaxAttempts(0)
+          .setInitialRetryDelayDuration(Duration.ofMillis(10))
+          .setMaxRetryDelayDuration(Duration.ofMillis(1000))
+          .build();
+
   private final RequestIdCreator requestIdCreator = new RequestIdCreatorImpl();
   private boolean rpcIsClosed;
   private final SpannerStub spannerStub;
@@ -452,8 +461,14 @@ public GapicSpannerRpc(final SpannerOptions options) {
         DIRECTPATH_CHANNEL_CREATED =
             ((GrpcTransportChannel) clientContext.getTransportChannel()).isDirectPath()
                 && isEnableDirectAccess;
-        this.readRetrySettings =
+        SpannerStubSettings.Builder defaultStubSettings = SpannerStubSettings.newBuilder();
+        RetrySettings configuredReadRetrySettings =
             options.getSpannerStubSettings().streamingReadSettings().getRetrySettings();
+        this.readRetrySettings =
+            configuredReadRetrySettings.equals(
+                    defaultStubSettings.streamingReadSettings().getRetrySettings())
+                ? DEFAULT_STREAMING_RETRY_SETTINGS
+                : configuredReadRetrySettings;
         Set streamingReadRetryableCodes =
             options.getSpannerStubSettings().streamingReadSettings().getRetryableCodes();
         this.readRetryableCodes =
@@ -463,8 +478,13 @@ public GapicSpannerRpc(final SpannerOptions options) {
                     .add(Code.RESOURCE_EXHAUSTED)
                     .build()
                 : streamingReadRetryableCodes;
-        this.executeQueryRetrySettings =
+        RetrySettings configuredQueryRetrySettings =
             options.getSpannerStubSettings().executeStreamingSqlSettings().getRetrySettings();
+        this.executeQueryRetrySettings =
+            configuredQueryRetrySettings.equals(
+                    defaultStubSettings.executeStreamingSqlSettings().getRetrySettings())
+                ? DEFAULT_STREAMING_RETRY_SETTINGS
+                : configuredQueryRetrySettings;
         Set executeStreamingSqlRetryableCodes =
             options.getSpannerStubSettings().executeStreamingSqlSettings().getRetryableCodes();
         this.executeQueryRetryableCodes =
diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java
index 807fd4be51c3..a0f18431315f 100644
--- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java
+++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ResumableStreamIteratorTest.java
@@ -23,7 +23,6 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-import com.google.api.client.util.BackOff;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.spanner.ErrorHandler.DefaultErrorHandler;
 import com.google.cloud.spanner.XGoogSpannerRequestId.NoopRequestIdCreator;
@@ -44,17 +43,14 @@
 import io.opencensus.trace.Span;
 import io.opencensus.trace.Tracing;
 import io.opentelemetry.api.OpenTelemetry;
-import io.opentelemetry.context.Scope;
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.lang.reflect.Method;
+import io.opentelemetry.api.common.Attributes;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
 import javax.annotation.Nullable;
-import org.junit.Assume;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -153,6 +149,7 @@ public boolean isLastStatement() {
 
   Starter starter = Mockito.mock(Starter.class);
   ResumableStreamIterator resumableStreamIterator;
+  private LongSupplier nanoTime = System::nanoTime;
 
   @Before
   public void setUp() {
@@ -164,20 +161,36 @@ public void setUp() {
   private void initWithLimit(int maxBufferSize) {
     initWithLimitAndRetrySettings(
         maxBufferSize,
-        SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetrySettings());
+        RetrySettings.newBuilder()
+            .setInitialRetryDelayDuration(java.time.Duration.ofMillis(10))
+            .setMaxRetryDelayDuration(java.time.Duration.ofMillis(1000))
+            .build());
   }
 
   private void initWithLimitAndRetrySettings(int maxBufferSize, RetrySettings retrySettings) {
+    initWithLimitAndRetrySettings(
+        maxBufferSize,
+        retrySettings,
+        new TraceWrapper(Tracing.getTracer(), OpenTelemetry.noop().getTracer(""), false));
+  }
+
+  private void initWithLimitAndRetrySettings(
+      int maxBufferSize, RetrySettings retrySettings, TraceWrapper tracer) {
     resumableStreamIterator =
         new ResumableStreamIterator(
             maxBufferSize,
             "",
             new OpenTelemetrySpan(mock(io.opentelemetry.api.trace.Span.class)),
-            new TraceWrapper(Tracing.getTracer(), OpenTelemetry.noop().getTracer(""), false),
+            tracer,
             DefaultErrorHandler.INSTANCE,
             retrySettings,
             SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetryableCodes(),
             NoopRequestIdCreator.INSTANCE) {
+          @Override
+          long nanoTime() {
+            return nanoTime.getAsLong();
+          }
+
           @Override
           AbstractResultSet.CloseableIterator startStream(
               @Nullable ByteString resumeToken,
@@ -203,14 +216,14 @@ public void simple() {
   public void closedOTSpan() {
     SpannerOptions.resetActiveTracingFramework();
     SpannerOptions.enableOpenTelemetryTraces();
-    Assume.assumeTrue(
-        "This test is only supported on JDK11 and lower",
-        JavaVersionUtil.getJavaMajorVersion() < 12);
 
     io.opentelemetry.api.trace.Span oTspan = mock(io.opentelemetry.api.trace.Span.class);
     ISpan span = new OpenTelemetrySpan(oTspan);
-    when(oTspan.makeCurrent()).thenReturn(mock(Scope.class));
-    setInternalState(ResumableStreamIterator.class, this.resumableStreamIterator, "span", span);
+    TraceWrapper tracer = mock(TraceWrapper.class);
+    when(tracer.spanBuilderWithExplicitParent(
+            Mockito.anyString(), Mockito.any(), Mockito.any(Attributes.class)))
+        .thenReturn(span);
+    initWithLimitAndRetrySettings(Integer.MAX_VALUE, RetrySettings.newBuilder().build(), tracer);
 
     ResultSetStream s1 = Mockito.mock(ResultSetStream.class);
     Mockito.when(starter.startStream(null, null)).thenReturn(new ResultSetIterator(s1));
@@ -228,12 +241,13 @@ public void closedOTSpan() {
   public void closedOCSpan() {
     SpannerOptions.resetActiveTracingFramework();
     SpannerOptions.enableOpenCensusTraces();
-    Assume.assumeTrue(
-        "This test is only supported on JDK11 and lower",
-        JavaVersionUtil.getJavaMajorVersion() < 12);
     Span mockSpan = mock(Span.class);
     ISpan span = new OpenCensusSpan(mockSpan);
-    setInternalState(ResumableStreamIterator.class, this.resumableStreamIterator, "span", span);
+    TraceWrapper tracer = mock(TraceWrapper.class);
+    when(tracer.spanBuilderWithExplicitParent(
+            Mockito.anyString(), Mockito.any(), Mockito.any(Attributes.class)))
+        .thenReturn(span);
+    initWithLimitAndRetrySettings(Integer.MAX_VALUE, RetrySettings.newBuilder().build(), tracer);
 
     ResultSetStream s1 = Mockito.mock(ResultSetStream.class);
     Mockito.when(starter.startStream(null, null)).thenReturn(new ResultSetIterator(s1));
@@ -311,18 +325,7 @@ public void restartWithHoldBackMidStream() {
   }
 
   @Test
-  public void retryableErrorWithoutRetryInfo() throws IOException {
-    Assume.assumeTrue(
-        "This test is only supported on JDK11 and lower",
-        JavaVersionUtil.getJavaMajorVersion() < 12);
-
-    BackOff backOff = mock(BackOff.class);
-    Mockito.when(backOff.nextBackOffMillis()).thenReturn(1L);
-    setInternalState(
-        ResumableStreamIterator.class, this.resumableStreamIterator, "backOff", backOff);
-
-    // The first stream fails before returning any resume token: receiving a new resume token
-    // resets the backoff by design, which would discard the injected mock backoff.
+  public void retryableErrorWithoutRetryInfo() {
     ResultSetStream s1 = Mockito.mock(ResultSetStream.class);
     Mockito.when(s1.next())
         .thenThrow(
@@ -339,7 +342,7 @@ public void retryableErrorWithoutRetryInfo() throws IOException {
         .thenReturn(new ResultSetIterator(s1))
         .thenReturn(new ResultSetIterator(s2));
     assertThat(consume(resumableStreamIterator)).containsExactly("a", "b").inOrder();
-    verify(backOff).nextBackOffMillis();
+    Mockito.verify(starter, Mockito.times(2)).startStream(null, null);
   }
 
   @Test(timeout = 60000L)
@@ -447,59 +450,95 @@ public void repeatedIdenticalResumeToken_doesNotResetAttempts() {
   }
 
   @Test
-  public void activeRetrySequence_preservesNegativeOneStartTime() throws Exception {
+  public void activeRetrySequence_preservesNegativeOneStartTime() {
     initWithLimitAndRetrySettings(
-        Integer.MAX_VALUE, RetrySettings.newBuilder().setMaxAttempts(2).build());
-    Field attempts = ResumableStreamIterator.class.getDeclaredField("attempts");
-    attempts.setAccessible(true);
-    attempts.setInt(resumableStreamIterator, 1);
-    Field startNanos = ResumableStreamIterator.class.getDeclaredField("retrySequenceStartNanos");
-    startNanos.setAccessible(true);
-    startNanos.setLong(resumableStreamIterator, -1L);
-    ResultSetStream stream = Mockito.mock(ResultSetStream.class);
-    Mockito.when(stream.next())
-        .thenThrow(new RetryableException(errorCodeParameter, "failed by test"));
-    Mockito.when(starter.startStream(null, null)).thenReturn(new ResultSetIterator(stream));
-
-    assertThrows(SpannerException.class, () -> consume(resumableStreamIterator));
-
-    assertEquals(-1L, startNanos.getLong(resumableStreamIterator));
+        Integer.MAX_VALUE,
+        RetrySettings.newBuilder()
+            .setTotalTimeoutDuration(java.time.Duration.ofSeconds(1))
+            .build());
+    SpannerException exception = new RetryableException(errorCodeParameter, "failed by test");
+    nanoTime = () -> -1L;
+    assertEquals(1L, resumableStreamIterator.checkRetryBudgetAndGetDelay(exception));
+    nanoTime = () -> TimeUnit.SECONDS.toNanos(1L);
+    assertThat(
+            assertThrows(
+                SpannerException.class,
+                () -> resumableStreamIterator.checkRetryBudgetAndGetDelay(exception)))
+        .isSameInstanceAs(exception);
   }
 
   @Test
-  public void totalTimeout_largeDelayDoesNotOverflow() throws Exception {
-    assertThat(totalTimeoutExceeded(1000L, -100L, Long.MAX_VALUE)).isTrue();
+  public void totalTimeout_largeDelayDoesNotOverflow() {
+    assertThat(totalTimeoutExceeded(1000L, 100L, Long.MAX_VALUE)).isTrue();
   }
 
   @Test
-  public void totalTimeout_negativeElapsedIsClamped() throws Exception {
-    assertThat(totalTimeoutExceeded(Long.MAX_VALUE, 60000L, Long.MAX_VALUE)).isTrue();
-    assertThat(totalTimeoutExceeded(Long.MAX_VALUE, 60000L, 0L)).isFalse();
+  public void totalTimeout_negativeElapsedIsClamped() {
+    assertThat(totalTimeoutExceeded(Long.MAX_VALUE, -60000L, Long.MAX_VALUE)).isTrue();
+    assertThat(totalTimeoutExceeded(Long.MAX_VALUE, -60000L, 0L)).isFalse();
   }
 
   @Test
-  public void totalTimeout_elapsedBudgetAndNegativeDelay() throws Exception {
-    assertThat(totalTimeoutExceeded(1000L, -60000L, -1L)).isTrue();
-    assertThat(totalTimeoutExceeded(1000L, 60000L, -1L)).isFalse();
-    assertThat(totalTimeoutExceeded(1000L, 60000L, 1000L)).isTrue();
+  public void totalTimeout_elapsedBudgetAndNegativeDelay() {
+    assertThat(totalTimeoutExceeded(1000L, 60000L, -2L)).isTrue();
+    assertThat(totalTimeoutExceeded(1000L, -60000L, -2L)).isFalse();
+    assertThat(totalTimeoutExceeded(1000L, -60000L, 1000L)).isTrue();
   }
 
-  private boolean totalTimeoutExceeded(long timeoutMillis, long startOffsetMillis, long delayMillis)
-      throws Exception {
+  private boolean totalTimeoutExceeded(long timeoutMillis, long elapsedMillis, long delayMillis) {
     initWithLimitAndRetrySettings(
         Integer.MAX_VALUE,
         RetrySettings.newBuilder()
             .setTotalTimeoutDuration(java.time.Duration.ofMillis(timeoutMillis))
             .build());
-    Field startNanos = ResumableStreamIterator.class.getDeclaredField("retrySequenceStartNanos");
-    startNanos.setAccessible(true);
-    startNanos.setLong(
-        resumableStreamIterator,
-        System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(startOffsetMillis));
-    Method totalTimeoutExceeded =
-        ResumableStreamIterator.class.getDeclaredMethod("totalTimeoutExceeded", long.class);
-    totalTimeoutExceeded.setAccessible(true);
-    return (boolean) totalTimeoutExceeded.invoke(resumableStreamIterator, delayMillis);
+    SpannerException exception =
+        Mockito.spy(new RetryableException(errorCodeParameter, "failed by test"));
+    when(exception.getRetryDelayInMillis()).thenReturn(0L);
+    nanoTime = () -> 0L;
+    assertEquals(0L, resumableStreamIterator.checkRetryBudgetAndGetDelay(exception));
+    nanoTime = () -> TimeUnit.MILLISECONDS.toNanos(elapsedMillis);
+    when(exception.getRetryDelayInMillis()).thenReturn(delayMillis);
+    try {
+      assertEquals(delayMillis, resumableStreamIterator.checkRetryBudgetAndGetDelay(exception));
+      return false;
+    } catch (SpannerException e) {
+      assertThat(e).isSameInstanceAs(exception);
+      return true;
+    }
+  }
+
+  @Test
+  public void retryBudget_usesConfiguredBackoffWithoutRetryInfo() {
+    initWithLimitAndRetrySettings(
+        Integer.MAX_VALUE,
+        RetrySettings.newBuilder()
+            .setInitialRetryDelayDuration(java.time.Duration.ofMillis(100))
+            .setMaxRetryDelayDuration(java.time.Duration.ofMillis(100))
+            .build());
+    SpannerException exception =
+        new RetryableException(
+            errorCodeParameter,
+            "failed by test",
+            errorCodeParameter.getGrpcStatus().asRuntimeException());
+    long delayMillis = resumableStreamIterator.checkRetryBudgetAndGetDelay(exception);
+    assertThat(delayMillis).isAtLeast(50L);
+    assertThat(delayMillis).isAtMost(150L);
+  }
+
+  @Test
+  public void rawGapicSettings_useConfiguredTimeoutWithoutSpecialCase() {
+    RetrySettings settings =
+        SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetrySettings();
+    initWithLimitAndRetrySettings(Integer.MAX_VALUE, settings);
+    SpannerException exception =
+        Mockito.spy(new RetryableException(errorCodeParameter, "failed by test"));
+    when(exception.getRetryDelayInMillis())
+        .thenReturn(settings.getTotalTimeoutDuration().toMillis());
+    assertThat(
+            assertThrows(
+                SpannerException.class,
+                () -> resumableStreamIterator.checkRetryBudgetAndGetDelay(exception)))
+        .isSameInstanceAs(exception);
   }
 
   @Test(timeout = 60000L)
@@ -758,18 +797,4 @@ static List consumeAtMost(int n, Iterator iterator) {
     }
     return r;
   }
-
-  /**
-   * Sets a private static final field to a specific value. This is only supported on Java11 and
-   * lower.
-   */
-  private static void setInternalState(Class c, Object target, String field, Object value) {
-    try {
-      Field f = c.getDeclaredField(field);
-      f.setAccessible(true);
-      f.set(target, value);
-    } catch (Exception e) {
-      throw new RuntimeException("Unable to set internal state on a private field.", e);
-    }
-  }
 }
diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java
index ec9f06d9ab90..747ee99b7a57 100644
--- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java
+++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java
@@ -30,6 +30,7 @@
 import com.google.protobuf.ListValue;
 import com.google.rpc.RetryInfo;
 import com.google.spanner.v1.ExecuteSqlRequest;
+import com.google.spanner.v1.ReadRequest;
 import com.google.spanner.v1.ResultSetMetadata;
 import com.google.spanner.v1.StructType;
 import com.google.spanner.v1.StructType.Field;
@@ -56,12 +57,14 @@
 /**
  * Tests that the streaming resume loop in {@link ResumableStreamIterator} honors the {@link
  * RetrySettings#getMaxAttempts()} and {@link RetrySettings#getTotalTimeout()} that have been
- * configured for ExecuteStreamingSql, instead of retrying indefinitely, while the default settings
- * (no maximum number of attempts) keep the existing unbounded resume behavior.
+ * configured for ExecuteStreamingSql and StreamingRead, instead of retrying indefinitely, while the
+ * default settings (no maximum number of attempts) keep the existing unbounded resume behavior.
  *
- * 
See https://github.com/googleapis/google-cloud-java/issues/12255. The tests in this class
- * drive a real {@link Spanner} client against an in-process mock Spanner server, so the entire
- * client stack (client -> GAX -> resume loop) is exercised.
+ * 
Setting {@code maxAttempts=1} is the supported way to disable streaming retries, as requested
+ * in https://github.com/googleapis/google-cloud-java/issues/12255; an empty retryable-code set
+ * alone does not disable retries for intrinsically retryable errors. The tests in this class drive
+ * a real {@link Spanner} client against an in-process mock Spanner server, so the entire client
+ * stack (client -> GAX -> resume loop) is exercised.
  */
 @RunWith(JUnit4.class)
 public class StreamingRetryBudgetMockServerTest {
@@ -113,6 +116,8 @@ private static Metadata createRetryInfoTrailers() {
   private DatabaseClient client;
   private Spanner spannerWithCustomRetrySettings;
   private DatabaseClient clientWithCustomRetrySettings;
+  private Spanner spannerWithoutRetries;
+  private DatabaseClient clientWithoutRetries;
   private Spanner spannerWithTotalTimeout;
   private DatabaseClient clientWithTotalTimeout;
 
@@ -136,6 +141,9 @@ public static void startStaticServer() throws Exception {
     mockSpanner = new MockSpannerServiceImpl();
     mockSpanner.setAbortProbability(0.0D); // We don't want any unpredictable aborted transactions.
     mockSpanner.putStatementResult(StatementResult.query(SELECT_QUERY, createResultSet(ROW_COUNT)));
+    mockSpanner.putStatementResult(
+        StatementResult.read(
+            "T", KeySet.all(), Collections.singletonList("C"), createResultSet(ROW_COUNT)));
 
     scheduledExecutor = new ScheduledThreadPoolExecutor(1);
     String uniqueName = InProcessServerBuilder.generateName();
@@ -209,6 +217,11 @@ public void setUp() {
         .executeStreamingSqlSettings()
         .setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED)
         .setRetrySettings(customRetrySettings);
+    builder
+        .getSpannerStubSettingsBuilder()
+        .streamingReadSettings()
+        .setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED)
+        .setRetrySettings(customRetrySettings);
     spannerWithCustomRetrySettings = builder.build().getService();
     clientWithCustomRetrySettings =
         spannerWithCustomRetrySettings.getDatabaseClient(
@@ -230,19 +243,112 @@ public void setUp() {
         .executeStreamingSqlSettings()
         .setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED)
         .setRetrySettings(totalTimeoutRetrySettings);
+    builder
+        .getSpannerStubSettingsBuilder()
+        .streamingReadSettings()
+        .setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED)
+        .setRetrySettings(totalTimeoutRetrySettings);
     spannerWithTotalTimeout = builder.build().getService();
     clientWithTotalTimeout =
         spannerWithTotalTimeout.getDatabaseClient(
             DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE]"));
+
+    // Leave retryable codes enabled in both layers: maxAttempts alone must disable retries.
+    RetrySettings noRetries =
+        customRetrySettings.toBuilder()
+            .setMaxAttempts(1)
+            .setTotalTimeoutDuration(Duration.ZERO)
+            .build();
+    builder
+        .getSpannerStubSettingsBuilder()
+        .executeStreamingSqlSettings()
+        .setRetrySettings(noRetries);
+    builder.getSpannerStubSettingsBuilder().streamingReadSettings().setRetrySettings(noRetries);
+    spannerWithoutRetries = builder.build().getService();
+    clientWithoutRetries =
+        spannerWithoutRetries.getDatabaseClient(
+            DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE]"));
   }
 
   @After
   public void tearDown() {
+    spannerWithoutRetries.close();
     spannerWithTotalTimeout.close();
     spannerWithCustomRetrySettings.close();
     spanner.close();
   }
 
+  @Test
+  public void defaultStreamingRetrySettings_areNormalized() {
+    RetrySettings expected =
+        RetrySettings.newBuilder()
+            .setTotalTimeoutDuration(Duration.ZERO)
+            .setMaxAttempts(0)
+            .setInitialRetryDelayDuration(Duration.ofMillis(10))
+            .setMaxRetryDelayDuration(Duration.ofMillis(1000))
+            .build();
+    assertEquals(expected, spanner.getOptions().getSpannerRpcV1().getExecuteQueryRetrySettings());
+    assertEquals(expected, spanner.getOptions().getSpannerRpcV1().getReadRetrySettings());
+  }
+
+  @Test(timeout = 60000L)
+  public void maxAttemptsOne_disablesStreamingSqlRetries() {
+    mockSpanner.setExecuteStreamingSqlExecutionTime(
+        SimulatedExecutionTime.ofStickyException(UNAVAILABLE));
+    mockSpanner.clearRequests();
+    try (ResultSet resultSet = clientWithoutRetries.singleUse().executeQuery(SELECT_QUERY)) {
+      SpannerException exception = assertThrows(SpannerException.class, resultSet::next);
+      assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode());
+    }
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+  }
+
+  @Test(timeout = 60000L)
+  public void maxAttemptsOne_disablesStreamingReadRetries() {
+    mockSpanner.setStreamingReadExecutionTime(
+        SimulatedExecutionTime.ofStickyException(UNAVAILABLE));
+    mockSpanner.clearRequests();
+    try (ResultSet resultSet =
+        clientWithoutRetries.singleUse().read("T", KeySet.all(), Collections.singletonList("C"))) {
+      SpannerException exception = assertThrows(SpannerException.class, resultSet::next);
+      assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode());
+    }
+    assertEquals(1, mockSpanner.countRequestsOfType(ReadRequest.class));
+  }
+
+  @Test(timeout = 60000L)
+  public void maxAttemptsExhausted_stopsStreamingReadRetries() {
+    mockSpanner.setStreamingReadExecutionTime(
+        SimulatedExecutionTime.ofStickyException(DEADLINE_EXCEEDED));
+    mockSpanner.clearRequests();
+    try (ResultSet resultSet =
+        clientWithCustomRetrySettings
+            .singleUse()
+            .read("T", KeySet.all(), Collections.singletonList("C"))) {
+      SpannerException exception = assertThrows(SpannerException.class, resultSet::next);
+      assertEquals(ErrorCode.DEADLINE_EXCEEDED, exception.getErrorCode());
+    }
+    // Both GAX and the resume loop allow two attempts, as with ExecuteStreamingSql.
+    assertEquals(4, mockSpanner.countRequestsOfType(ReadRequest.class));
+  }
+
+  @Test(timeout = 60000L)
+  public void totalTimeoutExhausted_stopsStreamingReadRetries() {
+    mockSpanner.setStreamingReadExecutionTime(
+        SimulatedExecutionTime.ofStickyException(UNAVAILABLE));
+    mockSpanner.clearRequests();
+    Stopwatch stopwatch = Stopwatch.createStarted();
+    try (ResultSet resultSet =
+        clientWithTotalTimeout
+            .singleUse()
+            .read("T", KeySet.all(), Collections.singletonList("C"))) {
+      SpannerException exception = assertThrows(SpannerException.class, resultSet::next);
+      assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode());
+    }
+    assertTrue(mockSpanner.countRequestsOfType(ReadRequest.class) > 1);
+    assertTrue(stopwatch.elapsed(TimeUnit.MILLISECONDS) < 10000L);
+  }
+
   /**
    * A query where every ExecuteStreamingSql attempt fails with DEADLINE_EXCEEDED must fail after
    * the configured maximum number of attempts. Without a bounded resume loop in {@link
@@ -338,6 +444,22 @@ public void defaultRetrySettings_keepsUnboundedResumes() {
     assertEquals(6, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
   }
 
+  @Test(timeout = 60000L)
+  public void defaultRetrySettings_keepsUnboundedReadResumes() {
+    mockSpanner.setStreamingReadExecutionTime(
+        SimulatedExecutionTime.ofExceptions(Collections.nCopies(5, UNAVAILABLE)));
+    mockSpanner.clearRequests();
+    int rows = 0;
+    try (ResultSet resultSet =
+        client.singleUse().read("T", KeySet.all(), Collections.singletonList("C"))) {
+      while (resultSet.next()) {
+        rows++;
+      }
+    }
+    assertEquals(ROW_COUNT, rows);
+    assertEquals(6, mockSpanner.countRequestsOfType(ReadRequest.class));
+  }
+
   /**
    * Only consecutive failures count against the maximum number of attempts: any progress on the
    * stream resets the budget. A stream that fails, resumes and makes progress, and then fails again
diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java
index d1e07caa5108..57fda236a3ec 100644
--- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java
+++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java
@@ -33,6 +33,7 @@
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.GrpcTransportChannel;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
+import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.ApiClientHeaderProvider;
 import com.google.api.gax.rpc.HeaderProvider;
@@ -671,6 +672,83 @@ public void testDynamicChannelPoolPrimesScaledUpChannelsWithSelectOne() throws E
     }
   }
 
+  @Test
+  public void testStreamingRetrySettingsNormalizedIndependently() {
+    for (boolean customizeRead : new boolean[] {true, false}) {
+      SpannerOptions.Builder builder = createSpannerOptions().toBuilder();
+      RetrySettings customSettings =
+          RetrySettings.newBuilder()
+              .setMaxAttempts(3)
+              .setTotalTimeoutDuration(Duration.ofSeconds(5))
+              .setInitialRetryDelayDuration(Duration.ofMillis(20))
+              .setMaxRetryDelayDuration(Duration.ofMillis(200))
+              .setRetryDelayMultiplier(2.0)
+              .build();
+      if (customizeRead) {
+        builder
+            .getSpannerStubSettingsBuilder()
+            .streamingReadSettings()
+            .setRetrySettings(customSettings);
+      } else {
+        builder
+            .getSpannerStubSettingsBuilder()
+            .executeStreamingSqlSettings()
+            .setRetrySettings(customSettings);
+      }
+      GapicSpannerRpc rpc = new GapicSpannerRpc(builder.build(), true);
+      try {
+        assertEquals(
+            customizeRead ? customSettings : GapicSpannerRpc.DEFAULT_STREAMING_RETRY_SETTINGS,
+            rpc.getReadRetrySettings());
+        assertEquals(
+            customizeRead ? GapicSpannerRpc.DEFAULT_STREAMING_RETRY_SETTINGS : customSettings,
+            rpc.getExecuteQueryRetrySettings());
+      } finally {
+        rpc.shutdown();
+      }
+    }
+  }
+
+  @Test
+  public void testCustomStreamingRetrySettingsRetainExplicitTimeout() {
+    for (boolean unlimited : new boolean[] {true, false}) {
+      SpannerOptions.Builder builder = createSpannerOptions().toBuilder();
+      RetrySettings.Builder readSettings =
+          builder
+              .getSpannerStubSettingsBuilder()
+              .streamingReadSettings()
+              .getRetrySettings()
+              .toBuilder()
+              .setMaxAttempts(2);
+      RetrySettings.Builder querySettings =
+          builder
+              .getSpannerStubSettingsBuilder()
+              .executeStreamingSqlSettings()
+              .getRetrySettings()
+              .toBuilder()
+              .setMaxAttempts(3);
+      if (unlimited) {
+        readSettings.setTotalTimeoutDuration(Duration.ZERO);
+        querySettings.setTotalTimeoutDuration(Duration.ZERO);
+      }
+      builder
+          .getSpannerStubSettingsBuilder()
+          .streamingReadSettings()
+          .setRetrySettings(readSettings.build());
+      builder
+          .getSpannerStubSettingsBuilder()
+          .executeStreamingSqlSettings()
+          .setRetrySettings(querySettings.build());
+      GapicSpannerRpc rpc = new GapicSpannerRpc(builder.build(), true);
+      try {
+        assertEquals(readSettings.build(), rpc.getReadRetrySettings());
+        assertEquals(querySettings.build(), rpc.getExecuteQueryRetrySettings());
+      } finally {
+        rpc.shutdown();
+      }
+    }
+  }
+
   @Test
   public void testCallCredentialsProviderPreferenceAboveCredentials() {
     SpannerOptions options =