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..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;
@@ -52,12 +50,28 @@
* 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.
+ *
+ * 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
+ * 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
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;
@@ -69,7 +83,20 @@ 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. Only meaningful when {@link #attempts} is nonzero. Used to enforce
+ * {@link RetrySettings#getTotalTimeout()} when a positive timeout is configured.
+ */
+ private long retrySequenceStartNanos = -1L;
+
private ByteString resumeToken;
private boolean finished;
private final XGoogSpannerRequestId requestId;
@@ -123,17 +150,42 @@ protected ResumableStreamIterator(
this.requestId = xGoogRequestIdCreator.nextRequestId(0);
}
- private ExponentialBackOff newBackOff() {
- if (Objects.equals(streamingRetrySettings, DEFAULT_STREAMING_RETRY_SETTINGS)) {
- return new ExponentialBackOff.Builder()
- .setMultiplier(streamingRetrySettings.getRetryDelayMultiplier())
- .setInitialIntervalMillis(
- Math.max(10, (int) streamingRetrySettings.getInitialRetryDelay().toMillis()))
- .setMaxIntervalMillis(
- Math.max(1000, (int) streamingRetrySettings.getMaxRetryDelay().toMillis()))
- .setMaxElapsedTimeMillis(Integer.MAX_VALUE) // Prevent Backoff.STOP from getting returned.
- .build();
+ /**
+ * 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 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) {
+ long totalTimeoutMillis = streamingRetrySettings.getTotalTimeout().toMillis();
+ if (totalTimeoutMillis <= 0L) {
+ return false;
+ }
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(nanoTime() - retrySequenceStartNanos);
+ if (elapsedMillis < 0L) {
+ elapsedMillis = 0L;
+ }
+ if (elapsedMillis >= totalTimeoutMillis) {
+ return true;
}
+ return Math.max(proposedDelayMillis, 0L) >= totalTimeoutMillis - elapsedMillis;
+ }
+
+ private ExponentialBackOff newBackOff() {
return new ExponentialBackOff.Builder()
.setMultiplier(streamingRetrySettings.getRetryDelayMultiplier())
// All of these values must be > 0.
@@ -150,19 +202,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 +310,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,25 +342,7 @@ protected PartialResultSet computeNext() {
}
} catch (SpannerException spannerException) {
if (safeToRetry && isRetryable(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)) {
- long delay = spannerException.getRetryDelayInMillis();
- if (delay != -1) {
- backoffSleep(context, delay);
- } else {
- if (this.backOff == null) {
- this.backOff = newBackOff();
- }
- backoffSleep(context, this.backOff);
- }
- }
+ handleRetryableException(context, spannerException);
continue;
}
@@ -331,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/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..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,10 +23,11 @@
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;
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;
@@ -42,16 +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 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;
@@ -150,6 +149,7 @@ public boolean isLastStatement() {
Starter starter = Mockito.mock(Starter.class);
ResumableStreamIterator resumableStreamIterator;
+ private LongSupplier nanoTime = System::nanoTime;
@Before
public void setUp() {
@@ -159,17 +159,38 @@ public void setUp() {
}
private void initWithLimit(int maxBufferSize) {
+ initWithLimitAndRetrySettings(
+ maxBufferSize,
+ 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,
- SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetrySettings(),
+ retrySettings,
SpannerStubSettings.newBuilder().executeStreamingSqlSettings().getRetryableCodes(),
NoopRequestIdCreator.INSTANCE) {
+ @Override
+ long nanoTime() {
+ return nanoTime.getAsLong();
+ }
+
@Override
AbstractResultSet.CloseableIterator startStream(
@Nullable ByteString resumeToken,
@@ -195,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));
@@ -220,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));
@@ -303,32 +325,288 @@ 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);
-
+ public void retryableErrorWithoutRetryInfo() {
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();
+ Mockito.verify(starter, Mockito.times(2)).startStream(null, null);
+ }
+
+ @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
+ public void activeRetrySequence_preservesNegativeOneStartTime() {
+ initWithLimitAndRetrySettings(
+ 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() {
+ assertThat(totalTimeoutExceeded(1000L, 100L, Long.MAX_VALUE)).isTrue();
+ }
+
+ @Test
+ 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() {
+ assertThat(totalTimeoutExceeded(1000L, 60000L, -2L)).isTrue();
+ assertThat(totalTimeoutExceeded(1000L, -60000L, -2L)).isFalse();
+ assertThat(totalTimeoutExceeded(1000L, -60000L, 1000L)).isTrue();
+ }
+
+ private boolean totalTimeoutExceeded(long timeoutMillis, long elapsedMillis, long delayMillis) {
+ initWithLimitAndRetrySettings(
+ Integer.MAX_VALUE,
+ RetrySettings.newBuilder()
+ .setTotalTimeoutDuration(java.time.Duration.ofMillis(timeoutMillis))
+ .build());
+ 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)
+ 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
@@ -519,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
new file mode 100644
index 000000000000..747ee99b7a57
--- /dev/null
+++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingRetryBudgetMockServerTest.java
@@ -0,0 +1,489 @@
+/*
+ * 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.ReadRequest;
+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 and StreamingRead, instead of retrying indefinitely, while the
+ * default settings (no maximum number of attempts) keep the existing unbounded resume behavior.
+ *
+ * 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 {
+ 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 ScheduledThreadPoolExecutor scheduledExecutor;
+ private static LocalChannelProvider channelProvider;
+
+ private Spanner spanner;
+ private DatabaseClient client;
+ private Spanner spannerWithCustomRetrySettings;
+ private DatabaseClient clientWithCustomRetrySettings;
+ private Spanner spannerWithoutRetries;
+ private DatabaseClient clientWithoutRetries;
+ 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)));
+ mockSpanner.putStatementResult(
+ StatementResult.read(
+ "T", KeySet.all(), Collections.singletonList("C"), 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(scheduledExecutor)
+ .addService(mockSpanner)
+ .build()
+ .start();
+ channelProvider = LocalChannelProvider.create(uniqueName);
+ }
+
+ @AfterClass
+ public static void stopServer() throws InterruptedException {
+ if (server != null) {
+ server.shutdown();
+ server.awaitTermination();
+ }
+ if (scheduledExecutor != null) {
+ scheduledExecutor.shutdown();
+ if (!scheduledExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
+ scheduledExecutor.shutdownNow();
+ }
+ }
+ }
+
+ @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);
+ builder
+ .getSpannerStubSettingsBuilder()
+ .streamingReadSettings()
+ .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);
+ 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
+ * 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));
+ }
+
+ @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
+ * 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));
+ }
+}
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 =