Skip to content

Commit 9695ace

Browse files
yifanzyifanzgcf-owl-bot[bot]rajatbhatta
authored
feat: Leader Aware Routing (#2214)
* feat: Add `x-goog-spanner-route-to-leader` header to Spanner RPC contexts for RW/PDML transactions. The header is added to support leader-aware-routing feature, which aims at reducing cross-regional latency for RW/PDML transactions in a multi-region instance. * feat: Add knob in SpannerOptions to allow users to opt out leader aware routing feature * fix: fix broken tests due to the merge * fix: resolve comments. * fix: resolve comments and add new tests to verify that the route-to-leader header exists for RW transactions and does not exist for RO transactions or when the leader aware routing feature is disabled. * fix: Update comments for SpannerOptions.disableLeaderAwareRouting * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: Disable leader aware routing by default for public preview --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com> Co-authored-by: Rajat Bhatta <[email protected]>
1 parent 85213c8 commit 9695ace

18 files changed

+674
-269
lines changed

‎google-cloud-spanner/clirr-ignored-differences.xml‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -222,6 +222,105 @@
222222
<className>com/google/cloud/spanner/connection/Connection</className>
223223
<method>com.google.cloud.spanner.ResultSet analyzeUpdateStatement(com.google.cloud.spanner.Statement, com.google.cloud.spanner.ReadContext$QueryAnalyzeMode, com.google.cloud.spanner.Options$UpdateOption[])</method>
224224
</difference>
225+
<difference>
226+
<differenceType>7004</differenceType>
227+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
228+
<method>com.google.spanner.v1.Transaction beginTransaction(com.google.spanner.v1.BeginTransactionRequest, java.util.Map)</method>
229+
</difference>
230+
<difference>
231+
<differenceType>7004</differenceType>
232+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
233+
<method>com.google.api.core.ApiFuture beginTransactionAsync(com.google.spanner.v1.BeginTransactionRequest, java.util.Map)</method>
234+
</difference>
235+
<difference>
236+
<differenceType>7004</differenceType>
237+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
238+
<method>com.google.api.core.ApiFuture beginTransactionAsync(com.google.spanner.v1.BeginTransactionRequest, java.util.Map)</method>
239+
</difference>
240+
<difference>
241+
<differenceType>7004</differenceType>
242+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
243+
<method>com.google.spanner.v1.ResultSet executeQuery(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map)</method>
244+
<to>com.google.spanner.v1.ResultSet executeQuery(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map, boolean)</to>
245+
</difference>
246+
<difference>
247+
<differenceType>7004</differenceType>
248+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
249+
<method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
250+
<to>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map, boolean)</to>
251+
</difference>
252+
<difference>
253+
<differenceType>7004</differenceType>
254+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
255+
<method>com.google.api.core.ApiFuture executeQueryAsync(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map)</method>
256+
</difference>
257+
<difference>
258+
<differenceType>7004</differenceType>
259+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
260+
<method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall read(com.google.spanner.v1.ReadRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
261+
</difference>
262+
<difference>
263+
<differenceType>7005</differenceType>
264+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
265+
<method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
266+
<to>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map, boolean)</to>
267+
</difference>
268+
<difference>
269+
<differenceType>7006</differenceType>
270+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
271+
<method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
272+
<to>com.google.spanner.v1.ResultSet</to>
273+
</difference>
274+
<difference>
275+
<differenceType>7006</differenceType>
276+
<className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
277+
<method>com.google.spanner.v1.ResultSet executeQuery(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map)</method>
278+
<to>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall</to>
279+
</difference>
280+
<difference>
281+
<differenceType>7004</differenceType>
282+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
283+
<method>com.google.spanner.v1.Transaction beginTransaction(com.google.spanner.v1.BeginTransactionRequest, java.util.Map)</method>
284+
</difference>
285+
<difference>
286+
<differenceType>7004</differenceType>
287+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
288+
<method>com.google.api.core.ApiFuture beginTransactionAsync(com.google.spanner.v1.BeginTransactionRequest, java.util.Map)</method>
289+
</difference>
290+
<difference>
291+
<differenceType>7004</differenceType>
292+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
293+
<method>com.google.spanner.v1.ResultSet executeQuery(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map)</method>
294+
</difference>
295+
<difference>
296+
<differenceType>7004</differenceType>
297+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
298+
<method>com.google.api.core.ApiFuture executeQueryAsync(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map)</method>
299+
</difference>
300+
<difference>
301+
<differenceType>7004</differenceType>
302+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
303+
<method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall read(com.google.spanner.v1.ReadRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
304+
</difference>
305+
<difference>
306+
<differenceType>7005</differenceType>
307+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
308+
<method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
309+
<to>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map, boolean)</to>
310+
</difference>
311+
<difference>
312+
<differenceType>7006</differenceType>
313+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
314+
<method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall executeQuery(com.google.spanner.v1.ExecuteSqlRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
315+
<to>com.google.spanner.v1.ResultSet</to>
316+
</difference>
317+
<difference>
318+
<differenceType>7006</differenceType>
319+
<className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
320+
<method>com.google.spanner.v1.ResultSet executeQuery(com.google.spanner.v1.ExecuteSqlRequest, java.util.Map)</method>
321+
<to>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall</to>
322+
</difference>
323+
225324
<!-- Savepoints -->
226325
<difference>
227326
<differenceType>7012</differenceType>

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

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,11 @@ private SingleReadContext(Builder builder) {
163163
this.bound = builder.bound;
164164
}
165165

166+
@Override
167+
protected boolean isRouteToLeader() {
168+
return false;
169+
}
170+
166171
@GuardedBy("lock")
167172
@Override
168173
void beforeReadOrQueryLocked() {
@@ -293,6 +298,11 @@ static Builder newBuilder() {
293298
}
294299
}
295300

301+
@Override
302+
protected boolean isRouteToLeader() {
303+
return false;
304+
}
305+
296306
@Override
297307
void beforeReadOrQuery() {
298308
super.beforeReadOrQuery();
@@ -347,7 +357,8 @@ void initTransaction() {
347357
.setSession(session.getName())
348358
.setOptions(options)
349359
.build();
350-
Transaction transaction = rpc.beginTransaction(request, session.getOptions());
360+
Transaction transaction =
361+
rpc.beginTransaction(request, session.getOptions(), isRouteToLeader());
351362
if (!transaction.hasReadTimestamp()) {
352363
throw SpannerExceptionFactory.newSpannerException(
353364
ErrorCode.INTERNAL, "Missing expected transaction.read_timestamp metadata field");
@@ -416,6 +427,10 @@ long getSeqNo() {
416427
return seqNo.incrementAndGet();
417428
}
418429

430+
protected boolean isRouteToLeader() {
431+
return false;
432+
}
433+
419434
@Override
420435
public final ResultSet read(
421436
String table, KeySet keys, Iterable<String> columns, ReadOption... options) {
@@ -667,7 +682,8 @@ CloseableIterator<PartialResultSet> startStream(@Nullable ByteString resumeToken
667682
request.setTransaction(selector);
668683
}
669684
SpannerRpc.StreamingCall call =
670-
rpc.executeQuery(request.build(), stream.consumer(), session.getOptions());
685+
rpc.executeQuery(
686+
request.build(), stream.consumer(), session.getOptions(), isRouteToLeader());
671687
call.request(prefetchChunks);
672688
stream.setCall(call, request.getTransaction().hasBegin());
673689
return stream;
@@ -798,7 +814,8 @@ CloseableIterator<PartialResultSet> startStream(@Nullable ByteString resumeToken
798814
}
799815
builder.setRequestOptions(buildRequestOptions(readOptions));
800816
SpannerRpc.StreamingCall call =
801-
rpc.read(builder.build(), stream.consumer(), session.getOptions());
817+
rpc.read(
818+
builder.build(), stream.consumer(), session.getOptions(), isRouteToLeader());
802819
call.request(prefetchChunks);
803820
stream.setCall(call, /* withBeginTransaction = */ builder.getTransaction().hasBegin());
804821
return stream;

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -202,7 +202,7 @@ private ByteString initTransaction() {
202202
TransactionOptions.newBuilder()
203203
.setPartitionedDml(TransactionOptions.PartitionedDml.getDefaultInstance()))
204204
.build();
205-
Transaction tx = rpc.beginTransaction(request, session.getOptions());
205+
Transaction tx = rpc.beginTransaction(request, session.getOptions(), true);
206206
if (tx.getId().isEmpty()) {
207207
throw SpannerExceptionFactory.newSpannerException(
208208
ErrorCode.INTERNAL,

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

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -275,7 +275,7 @@ public AsyncTransactionManagerImpl transactionManagerAsync(TransactionOption...
275275
@Override
276276
public void prepareReadWriteTransaction() {
277277
setActive(null);
278-
readyTransactionId = beginTransaction();
278+
readyTransactionId = beginTransaction(true);
279279
}
280280

281281
@Override
@@ -296,21 +296,21 @@ public void close() {
296296
}
297297
}
298298

299-
ByteString beginTransaction() {
299+
ByteString beginTransaction(boolean routeToLeader) {
300300
try {
301-
return beginTransactionAsync().get();
301+
return beginTransactionAsync(routeToLeader).get();
302302
} catch (ExecutionException e) {
303303
throw SpannerExceptionFactory.newSpannerException(e.getCause() == null ? e : e.getCause());
304304
} catch (InterruptedException e) {
305305
throw SpannerExceptionFactory.propagateInterrupt(e);
306306
}
307307
}
308308

309-
ApiFuture<ByteString> beginTransactionAsync() {
310-
return beginTransactionAsync(Options.fromTransactionOptions());
309+
ApiFuture<ByteString> beginTransactionAsync(boolean routeToLeader) {
310+
return beginTransactionAsync(Options.fromTransactionOptions(), routeToLeader);
311311
}
312312

313-
ApiFuture<ByteString> beginTransactionAsync(Options transactionOptions) {
313+
ApiFuture<ByteString> beginTransactionAsync(Options transactionOptions, boolean routeToLeader) {
314314
final SettableApiFuture<ByteString> res = SettableApiFuture.create();
315315
final Span span = tracer.spanBuilder(SpannerImpl.BEGIN_TRANSACTION).startSpan();
316316
final BeginTransactionRequest request =
@@ -319,7 +319,7 @@ ApiFuture<ByteString> beginTransactionAsync(Options transactionOptions) {
319319
.setOptions(createReadWriteTransactionOptions(transactionOptions))
320320
.build();
321321
final ApiFuture<Transaction> requestFuture =
322-
spanner.getRpc().beginTransactionAsync(request, options);
322+
spanner.getRpc().beginTransactionAsync(request, options, routeToLeader);
323323
requestFuture.addListener(
324324
tracer.withSpan(
325325
span,

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,7 @@ public class SpannerOptions extends ServiceOptions<Spanner, SpannerOptions> {
132132
private final CallCredentialsProvider callCredentialsProvider;
133133
private final CloseableExecutorProvider asyncExecutorProvider;
134134
private final String compressorName;
135+
private final boolean leaderAwareRoutingEnabled;
135136

136137
/**
137138
* Interface that can be used to provide {@link CallCredentials} instead of {@link Credentials} to
@@ -600,6 +601,7 @@ private SpannerOptions(Builder builder) {
600601
callCredentialsProvider = builder.callCredentialsProvider;
601602
asyncExecutorProvider = builder.asyncExecutorProvider;
602603
compressorName = builder.compressorName;
604+
leaderAwareRoutingEnabled = builder.leaderAwareRoutingEnabled;
603605
}
604606

605607
/**
@@ -700,6 +702,7 @@ public static class Builder
700702
private CloseableExecutorProvider asyncExecutorProvider;
701703
private String compressorName;
702704
private String emulatorHost = System.getenv("SPANNER_EMULATOR_HOST");
705+
private boolean leaderAwareRoutingEnabled = false;
703706

704707
private Builder() {
705708
// Manually set retry and polling settings that work.
@@ -1155,6 +1158,24 @@ public Builder setEmulatorHost(String emulatorHost) {
11551158
return this;
11561159
}
11571160

1161+
/**
1162+
* Enable leader aware routing. Leader aware routing would route all requests in RW/PDML
1163+
* transactions to the leader region.
1164+
*/
1165+
public Builder enableLeaderAwareRouting() {
1166+
this.leaderAwareRoutingEnabled = true;
1167+
return this;
1168+
}
1169+
1170+
/**
1171+
* Disable leader aware routing. Disabling leader aware routing would route all requests in
1172+
* RW/PDML transactions to any region.
1173+
*/
1174+
public Builder disableLeaderAwareRouting() {
1175+
this.leaderAwareRoutingEnabled = false;
1176+
return this;
1177+
}
1178+
11581179
@SuppressWarnings("rawtypes")
11591180
@Override
11601181
public SpannerOptions build() {
@@ -1291,6 +1312,10 @@ public String getCompressorName() {
12911312
return compressorName;
12921313
}
12931314

1315+
public boolean isLeaderAwareRoutingEnabled() {
1316+
return leaderAwareRoutingEnabled;
1317+
}
1318+
12941319
/** Returns the default query options to use for the specific database. */
12951320
public QueryOptions getDefaultQueryOptions(DatabaseId databaseId) {
12961321
// Use the specific query options for the database if any have been specified. These have

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

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,11 @@ private TransactionContextImpl(Builder builder) {
198198
this.finishedAsyncOperations.set(null);
199199
}
200200

201+
@Override
202+
protected boolean isRouteToLeader() {
203+
return true;
204+
}
205+
201206
private void increaseAsyncOperations() {
202207
synchronized (lock) {
203208
if (runningAsyncOperations == 0) {
@@ -255,7 +260,7 @@ ApiFuture<Void> ensureTxnAsync() {
255260

256261
private void createTxnAsync(final SettableApiFuture<Void> res) {
257262
span.addAnnotation("Creating Transaction");
258-
final ApiFuture<ByteString> fut = session.beginTransactionAsync(options);
263+
final ApiFuture<ByteString> fut = session.beginTransactionAsync(options, isRouteToLeader());
259264
fut.addListener(
260265
() -> {
261266
try {
@@ -717,7 +722,7 @@ private ResultSet internalExecuteUpdate(
717722
/* withTransactionSelector = */ true);
718723
try {
719724
com.google.spanner.v1.ResultSet resultSet =
720-
rpc.executeQuery(builder.build(), session.getOptions());
725+
rpc.executeQuery(builder.build(), session.getOptions(), isRouteToLeader());
721726
if (resultSet.getMetadata().hasTransaction()) {
722727
onTransactionMetadata(
723728
resultSet.getMetadata().getTransaction(), builder.getTransaction().hasBegin());
@@ -747,7 +752,7 @@ public ApiFuture<Long> executeUpdateAsync(Statement statement, UpdateOption... o
747752
// Register the update as an async operation that must finish before the transaction may
748753
// commit.
749754
increaseAsyncOperations();
750-
resultSet = rpc.executeQueryAsync(builder.build(), session.getOptions());
755+
resultSet = rpc.executeQueryAsync(builder.build(), session.getOptions(), isRouteToLeader());
751756
} catch (Throwable t) {
752757
decreaseAsyncOperations();
753758
throw t;

0 commit comments

Comments
 (0)