Skip to content
This repository was archived by the owner on Apr 7, 2026. It is now read-only.

Commit 5226c7f

Browse files
olavloiteolavloite
committed
fix: retry CreateSession also when waitForMinSessions is zero
The creation of a session at startup would only be retried if waitForMinSessions was set. This changed introduces retries for CreateSession at startup if the error code is potentially transient. The number of retries is set to maximum 10.
1 parent b92b5b3 commit 5226c7f

3 files changed

Lines changed: 345 additions & 3 deletions

File tree

‎google-cloud-spanner/src/main/java/com/google/cloud/spanner/MultiplexedSessionDatabaseClient.java‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -226,14 +226,18 @@ public void close() {
226226
final SettableApiFuture<SessionReference> initialSessionReferenceFuture =
227227
SettableApiFuture.create();
228228
this.multiplexedSessionReference = new AtomicReference<>(initialSessionReferenceFuture);
229-
asyncCreateMultiplexedSession(initialSessionReferenceFuture);
229+
230+
Duration waitDuration =
231+
sessionClient.getSpanner().getOptions().getSessionPoolOptions().getWaitForMinSessions();
232+
int initialAttempts = waitDuration == null || waitDuration.isZero() ? 10 : 1;
233+
asyncCreateMultiplexedSession(initialSessionReferenceFuture, initialAttempts);
230234
maybeWaitForSessionCreation(
231235
sessionClient.getSpanner().getOptions().getSessionPoolOptions(),
232236
initialSessionReferenceFuture);
233237
}
234238

235239
private void asyncCreateMultiplexedSession(
236-
SettableApiFuture<SessionReference> sessionReferenceFuture) {
240+
SettableApiFuture<SessionReference> sessionReferenceFuture, int remainingAttempts) {
237241
this.sessionClient.asyncCreateMultiplexedSession(
238242
new SessionConsumer() {
239243
@Override
@@ -263,7 +267,15 @@ public void onSessionCreateFailure(Throwable t, int createFailureForSessionCount
263267
MultiplexedSessionDatabaseClient.this.resourceNotFoundException.set(
264268
(ResourceNotFoundException) spannerException);
265269
}
270+
// Set the exception to trigger an error for all waiters.
271+
// Then retry the session creation if the error is (potentially) transient.
266272
sessionReferenceFuture.setException(t);
273+
if (remainingAttempts > 1
274+
&& RETRYABLE_ERROR_CODES.contains(spannerException.getErrorCode())) {
275+
final SettableApiFuture<SessionReference> future = SettableApiFuture.create();
276+
MultiplexedSessionDatabaseClient.this.multiplexedSessionReference.set(future);
277+
asyncCreateMultiplexedSession(future, remainingAttempts - 1);
278+
}
267279
}
268280
});
269281
}
@@ -283,7 +295,7 @@ private void maybeWaitForSessionCreation(
283295
// If any exception is thrown, then retry the multiplexed session creation
284296
if (sessionReferenceFuture == null) {
285297
sessionReferenceFuture = SettableApiFuture.create();
286-
asyncCreateMultiplexedSession(sessionReferenceFuture);
298+
asyncCreateMultiplexedSession(sessionReferenceFuture, 1);
287299
this.multiplexedSessionReference.set(sessionReferenceFuture);
288300
}
289301
try {

‎google-cloud-spanner/src/test/java/com/google/cloud/spanner/ExcludeFromChangeStreamTest.java‎

Lines changed: 299 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,299 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.spanner;
18+
19+
import static org.junit.Assert.assertEquals;
20+
import static org.junit.Assert.assertNotNull;
21+
import static org.junit.Assert.assertTrue;
22+
23+
import com.google.cloud.NoCredentials;
24+
import com.google.cloud.spanner.MockSpannerServiceImpl.SimulatedExecutionTime;
25+
import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
26+
import com.google.cloud.spanner.connection.AbstractMockServerTest;
27+
import com.google.cloud.spanner.connection.RandomResultSetGenerator;
28+
import com.google.common.collect.ImmutableList;
29+
import com.google.spanner.v1.BeginTransactionRequest;
30+
import com.google.spanner.v1.CommitRequest;
31+
import com.google.spanner.v1.ReadRequest;
32+
import io.grpc.ManagedChannelBuilder;
33+
import io.grpc.Status;
34+
import java.util.concurrent.atomic.AtomicBoolean;
35+
import org.junit.BeforeClass;
36+
import org.junit.Test;
37+
import org.junit.runner.RunWith;
38+
import org.junit.runners.JUnit4;
39+
40+
@RunWith(JUnit4.class)
41+
public class ExcludeFromChangeStreamTest extends AbstractMockServerTest {
42+
43+
@BeforeClass
44+
public static void setupReadResult() {
45+
RandomResultSetGenerator generator = new RandomResultSetGenerator(10);
46+
mockSpanner.putStatementResult(
47+
StatementResult.query(
48+
Statement.of("SELECT my-column FROM my-table WHERE 1=1"), generator.generate()));
49+
}
50+
51+
private Spanner createSpanner() {
52+
return SpannerOptions.newBuilder()
53+
.setProjectId("fake-project")
54+
.setHost("http://localhost:" + getPort())
55+
.setCredentials(NoCredentials.getInstance())
56+
.setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
57+
.build()
58+
.getService();
59+
}
60+
61+
@Test
62+
public void testStandardTransaction() {
63+
try (Spanner spanner = createSpanner()) {
64+
for (int i = 0; i < 10; i++) {
65+
DatabaseClient client =
66+
spanner.getDatabaseClient(
67+
DatabaseId.of("fake-project", "fake-instance", "fake-database"));
68+
client
69+
.readWriteTransaction(Options.tag("some-tag"), Options.excludeTxnFromChangeStreams())
70+
.run(
71+
transaction -> {
72+
try (ResultSet resultSet =
73+
transaction.read("my-table", KeySet.all(), ImmutableList.of("my-column"))) {
74+
while (resultSet.next()) {}
75+
}
76+
transaction.buffer(
77+
Mutation.newInsertOrUpdateBuilder("my-table")
78+
.set("my-column")
79+
.to(1L)
80+
.build());
81+
return null;
82+
});
83+
assertEquals(0, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
84+
assertEquals(1, mockSpanner.countRequestsOfType(ReadRequest.class));
85+
assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
86+
87+
ReadRequest readRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(0);
88+
assertTrue(readRequest.hasTransaction());
89+
assertTrue(readRequest.getTransaction().hasBegin());
90+
assertTrue(readRequest.getTransaction().getBegin().hasReadWrite());
91+
assertTrue(readRequest.getTransaction().getBegin().getExcludeTxnFromChangeStreams());
92+
93+
CommitRequest commitRequest = mockSpanner.getRequestsOfType(CommitRequest.class).get(0);
94+
assertNotNull(commitRequest.getTransactionId());
95+
96+
mockSpanner.clearRequests();
97+
}
98+
}
99+
}
100+
101+
@Test
102+
public void testTransactionAbortedDuringRead() {
103+
try (Spanner spanner = createSpanner()) {
104+
for (int i = 0; i < 10; i++) {
105+
DatabaseClient client =
106+
spanner.getDatabaseClient(
107+
DatabaseId.of("fake-project", "fake-instance", "fake-database"));
108+
AtomicBoolean hasAborted = new AtomicBoolean(false);
109+
client
110+
.readWriteTransaction(Options.tag("some-tag"), Options.excludeTxnFromChangeStreams())
111+
.run(
112+
transaction -> {
113+
if (hasAborted.compareAndSet(false, true)) {
114+
mockSpanner.abortNextStatement();
115+
}
116+
try (ResultSet resultSet =
117+
transaction.read("my-table", KeySet.all(), ImmutableList.of("my-column"))) {
118+
while (resultSet.next()) {}
119+
}
120+
transaction.buffer(
121+
Mutation.newInsertOrUpdateBuilder("my-table")
122+
.set("my-column")
123+
.to(1L)
124+
.build());
125+
return null;
126+
});
127+
assertEquals(1, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
128+
assertEquals(2, mockSpanner.countRequestsOfType(ReadRequest.class));
129+
assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
130+
131+
BeginTransactionRequest beginRequest =
132+
mockSpanner.getRequestsOfType(BeginTransactionRequest.class).get(0);
133+
assertTrue(beginRequest.getOptions().hasReadWrite());
134+
assertTrue(beginRequest.getOptions().getExcludeTxnFromChangeStreams());
135+
136+
ReadRequest firstReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(0);
137+
assertTrue(firstReadRequest.hasTransaction());
138+
assertTrue(firstReadRequest.getTransaction().hasBegin());
139+
assertTrue(firstReadRequest.getTransaction().getBegin().hasReadWrite());
140+
assertTrue(firstReadRequest.getTransaction().getBegin().getExcludeTxnFromChangeStreams());
141+
142+
ReadRequest secondReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(1);
143+
assertTrue(secondReadRequest.hasTransaction());
144+
assertTrue(secondReadRequest.getTransaction().hasId());
145+
146+
CommitRequest commitRequest = mockSpanner.getRequestsOfType(CommitRequest.class).get(0);
147+
assertNotNull(commitRequest.getTransactionId());
148+
149+
mockSpanner.clearRequests();
150+
}
151+
}
152+
}
153+
154+
@Test
155+
public void testTransactionAbortedDuringCommit() {
156+
try (Spanner spanner = createSpanner()) {
157+
for (int i = 0; i < 10; i++) {
158+
DatabaseClient client =
159+
spanner.getDatabaseClient(
160+
DatabaseId.of("fake-project", "fake-instance", "fake-database"));
161+
AtomicBoolean hasAborted = new AtomicBoolean(false);
162+
client
163+
.readWriteTransaction(Options.tag("some-tag"), Options.excludeTxnFromChangeStreams())
164+
.run(
165+
transaction -> {
166+
try (ResultSet resultSet =
167+
transaction.read("my-table", KeySet.all(), ImmutableList.of("my-column"))) {
168+
while (resultSet.next()) {}
169+
}
170+
if (hasAborted.compareAndSet(false, true)) {
171+
mockSpanner.abortNextStatement();
172+
}
173+
transaction.buffer(
174+
Mutation.newInsertOrUpdateBuilder("my-table")
175+
.set("my-column")
176+
.to(1L)
177+
.build());
178+
return null;
179+
});
180+
assertEquals(0, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
181+
assertEquals(2, mockSpanner.countRequestsOfType(ReadRequest.class));
182+
assertEquals(2, mockSpanner.countRequestsOfType(CommitRequest.class));
183+
184+
ReadRequest firstReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(0);
185+
assertTrue(firstReadRequest.hasTransaction());
186+
assertTrue(firstReadRequest.getTransaction().hasBegin());
187+
assertTrue(firstReadRequest.getTransaction().getBegin().hasReadWrite());
188+
assertTrue(firstReadRequest.getTransaction().getBegin().getExcludeTxnFromChangeStreams());
189+
190+
ReadRequest secondReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(1);
191+
assertTrue(secondReadRequest.hasTransaction());
192+
assertTrue(secondReadRequest.getTransaction().hasBegin());
193+
assertTrue(secondReadRequest.getTransaction().getBegin().hasReadWrite());
194+
assertTrue(secondReadRequest.getTransaction().getBegin().getExcludeTxnFromChangeStreams());
195+
196+
for (CommitRequest commitRequest : mockSpanner.getRequestsOfType(CommitRequest.class)) {
197+
assertNotNull(commitRequest.getTransactionId());
198+
}
199+
mockSpanner.clearRequests();
200+
}
201+
}
202+
}
203+
204+
@Test
205+
public void testReadReturnsUnavailable() {
206+
207+
try (Spanner spanner = createSpanner()) {
208+
for (int i = 0; i < 10; i++) {
209+
mockSpanner.setStreamingReadExecutionTime(
210+
SimulatedExecutionTime.ofException(Status.UNAVAILABLE.asRuntimeException()));
211+
DatabaseClient client =
212+
spanner.getDatabaseClient(
213+
DatabaseId.of("fake-project", "fake-instance", "fake-database"));
214+
client
215+
.readWriteTransaction(Options.tag("some-tag"), Options.excludeTxnFromChangeStreams())
216+
.run(
217+
transaction -> {
218+
try (ResultSet resultSet =
219+
transaction.read("my-table", KeySet.all(), ImmutableList.of("my-column"))) {
220+
while (resultSet.next()) {}
221+
}
222+
transaction.buffer(
223+
Mutation.newInsertOrUpdateBuilder("my-table")
224+
.set("my-column")
225+
.to(1L)
226+
.build());
227+
return null;
228+
});
229+
assertEquals(0, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
230+
assertEquals(2, mockSpanner.countRequestsOfType(ReadRequest.class));
231+
assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
232+
233+
ReadRequest firstReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(0);
234+
assertTrue(firstReadRequest.hasTransaction());
235+
assertTrue(firstReadRequest.getTransaction().hasBegin());
236+
assertTrue(firstReadRequest.getTransaction().getBegin().hasReadWrite());
237+
assertTrue(firstReadRequest.getTransaction().getBegin().getExcludeTxnFromChangeStreams());
238+
239+
ReadRequest secondReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(1);
240+
assertTrue(secondReadRequest.hasTransaction());
241+
assertTrue(secondReadRequest.getTransaction().hasBegin());
242+
assertTrue(secondReadRequest.getTransaction().getBegin().hasReadWrite());
243+
assertTrue(secondReadRequest.getTransaction().getBegin().getExcludeTxnFromChangeStreams());
244+
245+
CommitRequest commitRequest = mockSpanner.getRequestsOfType(CommitRequest.class).get(0);
246+
assertNotNull(commitRequest.getTransactionId());
247+
248+
mockSpanner.clearRequests();
249+
}
250+
}
251+
}
252+
253+
@Test
254+
public void testReadReturnsUnavailableHalfway() {
255+
try (Spanner spanner = createSpanner()) {
256+
for (int i = 0; i < 10; i++) {
257+
mockSpanner.setStreamingReadExecutionTime(
258+
SimulatedExecutionTime.ofStreamException(Status.UNAVAILABLE.asRuntimeException(), 2));
259+
260+
DatabaseClient client =
261+
spanner.getDatabaseClient(
262+
DatabaseId.of("fake-project", "fake-instance", "fake-database"));
263+
client
264+
.readWriteTransaction(Options.tag("some-tag"), Options.excludeTxnFromChangeStreams())
265+
.run(
266+
transaction -> {
267+
try (ResultSet resultSet =
268+
transaction.read("my-table", KeySet.all(), ImmutableList.of("my-column"))) {
269+
while (resultSet.next()) {}
270+
}
271+
transaction.buffer(
272+
Mutation.newInsertOrUpdateBuilder("my-table")
273+
.set("my-column")
274+
.to(1L)
275+
.build());
276+
return null;
277+
});
278+
assertEquals(0, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
279+
assertEquals(2, mockSpanner.countRequestsOfType(ReadRequest.class));
280+
assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
281+
282+
ReadRequest firstReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(0);
283+
assertTrue(firstReadRequest.hasTransaction());
284+
assertTrue(firstReadRequest.getTransaction().hasBegin());
285+
assertTrue(firstReadRequest.getTransaction().getBegin().hasReadWrite());
286+
assertTrue(firstReadRequest.getTransaction().getBegin().getExcludeTxnFromChangeStreams());
287+
288+
ReadRequest secondReadRequest = mockSpanner.getRequestsOfType(ReadRequest.class).get(1);
289+
assertTrue(secondReadRequest.hasTransaction());
290+
assertTrue(secondReadRequest.getTransaction().hasId());
291+
292+
CommitRequest commitRequest = mockSpanner.getRequestsOfType(CommitRequest.class).get(0);
293+
assertNotNull(commitRequest.getTransactionId());
294+
295+
mockSpanner.clearRequests();
296+
}
297+
}
298+
}
299+
}

‎google-cloud-spanner/src/test/java/com/google/cloud/spanner/MultiplexedSessionDatabaseClientMockServerTest.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,37 @@ public void createSpannerInstance() {
101101
.getService();
102102
}
103103

104+
@Test
105+
public void testCreateSessionDeadlineExceeded() {
106+
// Simulate a problem with the CreateSession RPC making it slow.
107+
mockSpanner.setCreateSessionExecutionTime(
108+
SimulatedExecutionTime.ofException(Status.DEADLINE_EXCEEDED.asRuntimeException()));
109+
110+
Spanner testSpanner =
111+
SpannerOptions.newBuilder()
112+
.setProjectId("test-project")
113+
.setChannelProvider(channelProvider)
114+
.setCredentials(NoCredentials.getInstance())
115+
.build()
116+
.getService();
117+
DatabaseClient client = testSpanner.getDatabaseClient(DatabaseId.of("p", "i", "d"));
118+
119+
// The first attempt should lead to a DEADLINE_EXCEEDED error being propagated from the
120+
// CreateSession attempt.
121+
try (ResultSet resultSet = client.singleUse().executeQuery(STATEMENT)) {
122+
SpannerException exception = assertThrows(SpannerException.class, resultSet::next);
123+
assertEquals(ErrorCode.DEADLINE_EXCEEDED, exception.getErrorCode());
124+
}
125+
126+
// Remove the simulated problem on the mock server.
127+
// The next attempt should then succeed.
128+
mockSpanner.removeAllExecutionTimes();
129+
try (ResultSet resultSet = client.singleUse().executeQuery(STATEMENT)) {
130+
//noinspection StatementWithEmptyBody
131+
while (resultSet.next()) {}
132+
}
133+
}
134+
104135
@Test
105136
public void testMultiUseReadOnlyTransactionUsesSameSession() {
106137
// Execute two queries using the same transaction. Both queries should use the same

0 commit comments

Comments
 (0)