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

Added Resource Based Routing Support #40

Closed
mbril wants to merge 10 commits into from
Closed

Added Resource Based Routing Support #40

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
e640b13
Initial import of Resource Based Routing Impl from old repo
mbril Jan 22, 2020
44ee822
reorder
mbril Jan 22, 2020
aee6992
Applied formatting
mbril Jan 22, 2020
87d76fd
Merge branch 'master' of github.com:googleapis/java-spanner into mb.a…
mbril Jan 22, 2020
cf0d792
Moved InstanceConfigRpcCache to v1 package per comment
mbril Jan 22, 2020
db780c9
Added feature flag
mbril Jan 22, 2020
cee129c
Merge branch 'master' of github.com:googleapis/java-spanner into mb.a…
mbril Jan 29, 2020
37d1424
Adapted clirr
mbril Jan 29, 2020
0472136
Merge branch 'master' of github.com:googleapis/java-spanner into mb.a…
mbril Feb 1, 2020
8041e08
Addressed comments per spec
mbril Feb 3, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Applied formatting
  • Loading branch information
@mbril
mbril committed Jan 22, 2020
commit aee69925f8565cab048169801d48b4ef437bb590
1 change: 0 additions & 1 deletion google-cloud-spanner/src/main/java/com/google/cloud/spanner/Instance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import com.google.cloud.Policy;
import com.google.spanner.admin.database.v1.CreateDatabaseMetadata;
import com.google.spanner.admin.instance.v1.UpdateInstanceMetadata;

import java.util.List;
import java.util.Map;

Expand Down
98 changes: 49 additions & 49 deletions google-cloud-spanner/src/main/java/com/google/cloud/spanner/InstanceConfigRpcCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,59 +41,59 @@ public class InstanceConfigRpcCache {
public InstanceConfigRpcCache(final GapicSpannerRpc projectClient) {
this.projectClient = projectClient;
cache =
CacheBuilder.newBuilder()
.build(
new CacheLoader<InstanceName, SpannerRpc>() {
@Override
public SpannerRpc load(InstanceName instanceName) throws SpannerException {
GetInstanceRequest request =
GetInstanceRequest.newBuilder()
.setName(instanceName.toString())
.setFieldMask(FieldMask.newBuilder().addPaths("endpoint_urls"))
.build();
SpannerOptions.Builder optionsBuilder = projectClient.getOptions().toBuilder();
try {
Instance instance = projectClient.getInstance(request);
if (instance.getEndpointUrisCount() > 0) {
optionsBuilder.setHost(instance.getEndpointUris(0));
} else {
return projectClient;
}
} catch (SpannerException e) {
if (e.getErrorCode() == ErrorCode.UNIMPLEMENTED) {
// Ignore...
// This is backwards compatibility.
return projectClient;
} else {
logger.log(
Level.WARNING,
"Failed while resolving endpoint URLs for instance:"
+ instanceName.toString()
+ " reason: "
+ e.getErrorCode().name());
throw e;
}
} catch (Exception e) {
logger.log(
Level.WARNING,
"Failed while resolving endpoint URLs for instance:"
+ instanceName.toString());
throw SpannerExceptionFactory.newSpannerException(e);
}
return new GapicSpannerRpc(optionsBuilder.build(), false);
}
});
CacheBuilder.newBuilder()
.build(
new CacheLoader<InstanceName, SpannerRpc>() {
@Override
public SpannerRpc load(InstanceName instanceName) throws SpannerException {
GetInstanceRequest request =
GetInstanceRequest.newBuilder()
.setName(instanceName.toString())
.setFieldMask(FieldMask.newBuilder().addPaths("endpoint_urls"))
.build();
SpannerOptions.Builder optionsBuilder = projectClient.getOptions().toBuilder();
try {
Instance instance = projectClient.getInstance(request);
if (instance.getEndpointUrisCount() > 0) {
optionsBuilder.setHost(instance.getEndpointUris(0));
} else {
return projectClient;
}
} catch (SpannerException e) {
if (e.getErrorCode() == ErrorCode.UNIMPLEMENTED) {
// Ignore...
// This is backwards compatibility.
return projectClient;
} else {
logger.log(
Level.WARNING,
"Failed while resolving endpoint URLs for instance:"
+ instanceName.toString()
+ " reason: "
+ e.getErrorCode().name());
throw e;
}
} catch (Exception e) {
logger.log(
Level.WARNING,
"Failed while resolving endpoint URLs for instance:"
+ instanceName.toString());
throw SpannerExceptionFactory.newSpannerException(e);
}
return new GapicSpannerRpc(optionsBuilder.build(), false);
}
});
}

public SpannerRpc get(SessionName sessionName) {
InstanceName instanceName =
InstanceName.of(sessionName.getProject(), sessionName.getInstance());
InstanceName.of(sessionName.getProject(), sessionName.getInstance());
return get(instanceName);
}

public SpannerRpc get(DatabaseName databaseName) {
InstanceName instanceName =
InstanceName.of(databaseName.getProject(), databaseName.getInstance());
InstanceName.of(databaseName.getProject(), databaseName.getInstance());
return get(instanceName);
}

Expand All @@ -103,11 +103,11 @@ private SpannerRpc get(InstanceName instanceName) {
return spannerRpc;
} catch (ExecutionException e) {
logger.log(
Level.FINE, "Failed looking up instance in cache. id:" + instanceName.toString(), e);
Level.FINE, "Failed looking up instance in cache. id:" + instanceName.toString(), e);
throw SpannerExceptionFactory.newSpannerException(
ErrorCode.NOT_FOUND,
"Failed getting RPC Client for Instance:" + instanceName.toString(),
e);
ErrorCode.NOT_FOUND,
"Failed getting RPC Client for Instance:" + instanceName.toString(),
e);
}
}

Expand All @@ -118,4 +118,4 @@ public void invalidateAll() {
}
cache.invalidateAll();
}
}
}
2 changes: 0 additions & 2 deletions google-cloud-spanner/src/main/java/com/google/cloud/spanner/InstanceInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.FieldMask;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
Expand Down Expand Up @@ -174,7 +173,6 @@ public InstanceInfo build() {
private final ImmutableMap<String, String> labels;
private final ImmutableList<String> endpointUris;


InstanceInfo(BuilderImpl builder) {
this.id = builder.id;
this.configId = builder.configId;
Expand Down
2 changes: 1 addition & 1 deletion google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,7 @@ SessionImpl createSession() {
try (Scope s = SpannerImpl.tracer.withSpan(span)) {
SpannerRpc spannerRpc = spanner.getRpc(DatabaseName.parse(db.getName()));
com.google.spanner.v1.Session session =
spannerRpc.createSession(db.getName(), spanner.getOptions().getSessionLabels(), options);
spannerRpc.createSession(db.getName(), spanner.getOptions().getSessionLabels(), options);
span.end();
return new SessionImpl(spanner, session.getName(), options);
} catch (RuntimeException e) {
Expand Down
14 changes: 9 additions & 5 deletions google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ public String getName() {
public long executePartitionedUpdate(Statement stmt) {
setActive(null);
PartitionedDMLTransaction txn =
new PartitionedDMLTransaction(this, spanner.getRpc(sessionName));
new PartitionedDMLTransaction(this, spanner.getRpc(sessionName));
return txn.executePartitionedUpdate(stmt, spanner.getOptions().getPartitionedDmlTimeout());
}

Expand Down Expand Up @@ -161,8 +161,8 @@ public ReadContext singleUse() {
@Override
public ReadContext singleUse(TimestampBound bound) {
return setActive(
new SingleReadContext(
this, bound, spanner.getRpc(sessionName), spanner.getDefaultPrefetchChunks()));
new SingleReadContext(
this, bound, spanner.getRpc(sessionName), spanner.getDefaultPrefetchChunks()));
}

@Override
Expand Down Expand Up @@ -192,7 +192,8 @@ public ReadOnlyTransaction readOnlyTransaction(TimestampBound bound) {
@Override
public TransactionRunner readWriteTransaction() {
return setActive(
new TransactionRunnerImpl(this, spanner.getRpc(sessionName), spanner.getDefaultPrefetchChunks()));
new TransactionRunnerImpl(
this, spanner.getRpc(sessionName), spanner.getDefaultPrefetchChunks()));
}

@Override
Expand Down Expand Up @@ -238,7 +239,10 @@ ByteString beginTransaction() {
TransactionContextImpl newTransaction() {
TransactionContextImpl txn =
new TransactionContextImpl(
this, readyTransactionId, spanner.getRpc(sessionName), spanner.getDefaultPrefetchChunks());
this,
readyTransactionId,
spanner.getRpc(sessionName),
spanner.getDefaultPrefetchChunks());
return txn;
}

Expand Down
1 change: 0 additions & 1 deletion google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
Expand Down
82 changes: 47 additions & 35 deletions google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -529,7 +529,7 @@ public List<Session> batchCreateSessions(
return get(getSpannerStub(DatabaseName.parse(databaseName))
.batchCreateSessionsCallable()
.futureCall(request, context))
.getSessionList();
.getSessionList();
}

@Override
Expand All @@ -545,27 +545,31 @@ public Session createSession(
CreateSessionRequest request = requestBuilder.build();
GrpcCallContext context = newCallContext(options, databaseName);
return get(
getSpannerStub(DatabaseName.parse(databaseName))
.createSessionCallable()
.futureCall(request, context)); }
getSpannerStub(DatabaseName.parse(databaseName))
.createSessionCallable()
.futureCall(request, context));
}

@Override
public void deleteSession(String sessionName, @Nullable Map<Option, ?> options)
throws SpannerException {
DeleteSessionRequest request = DeleteSessionRequest.newBuilder().setName(sessionName).build();
GrpcCallContext context = newCallContext(options, sessionName);
get(getSpannerStub(SessionName.parse(sessionName))
.deleteSessionCallable()
.futureCall(request, context)); }
get(
getSpannerStub(SessionName.parse(sessionName))
.deleteSessionCallable()
.futureCall(request, context));
}

@Override
public StreamingCall read(
ReadRequest request, ResultStreamConsumer consumer, @Nullable Map<Option, ?> options) {
GrpcCallContext context = newCallContext(options, request.getSession());
SpannerResponseObserver responseObserver = new SpannerResponseObserver(consumer);
getSpannerStub(SessionName.parse(request.getSession()))
.streamingReadCallable()
.call(request, responseObserver, context); final StreamController controller = responseObserver.getController();
.streamingReadCallable()
.call(request, responseObserver, context);
final StreamController controller = responseObserver.getController();
return new StreamingCall() {
@Override
public void request(int numMessage) {
Expand All @@ -585,27 +589,29 @@ public void cancel(String message) {
public ResultSet executeQuery(ExecuteSqlRequest request, @Nullable Map<Option, ?> options) {
GrpcCallContext context = newCallContext(options, request.getSession());
return get(
getSpannerStub(SessionName.parse(request.getSession()))
.executeSqlCallable()
.futureCall(request, context)); }
getSpannerStub(SessionName.parse(request.getSession()))
.executeSqlCallable()
.futureCall(request, context));
}

@Override
public ResultSet executePartitionedDml(
ExecuteSqlRequest request, @Nullable Map<Option, ?> options, Duration timeout) {
GrpcCallContext context = newCallContext(options, request.getSession(), timeout);
return get(
getSpannerStub(SessionName.parse(request.getSession()))
.executeSqlCallable()
.futureCall(request, context)); }
getSpannerStub(SessionName.parse(request.getSession()))
.executeSqlCallable()
.futureCall(request, context));
}

@Override
public StreamingCall executeQuery(
ExecuteSqlRequest request, ResultStreamConsumer consumer, @Nullable Map<Option, ?> options) {
GrpcCallContext context = newCallContext(options, request.getSession());
SpannerResponseObserver responseObserver = new SpannerResponseObserver(consumer);
getSpannerStub(SessionName.parse(request.getSession()))
.executeStreamingSqlCallable()
.call(request, responseObserver, context);
.executeStreamingSqlCallable()
.call(request, responseObserver, context);
final StreamController controller = responseObserver.getController();
return new StreamingCall() {
@Override
Expand All @@ -628,54 +634,60 @@ public ExecuteBatchDmlResponse executeBatchDml(

GrpcCallContext context = newCallContext(options, request.getSession());
return get(
getSpannerStub(SessionName.parse(request.getSession()))
.executeBatchDmlCallable()
.futureCall(request, context));
getSpannerStub(SessionName.parse(request.getSession()))
.executeBatchDmlCallable()
.futureCall(request, context));
}

@Override
public Transaction beginTransaction(
BeginTransactionRequest request, @Nullable Map<Option, ?> options) throws SpannerException {
GrpcCallContext context = newCallContext(options, request.getSession());
return get(
getSpannerStub(SessionName.parse(request.getSession()))
.beginTransactionCallable()
.futureCall(request, context)); }
getSpannerStub(SessionName.parse(request.getSession()))
.beginTransactionCallable()
.futureCall(request, context));
}

@Override
public CommitResponse commit(CommitRequest commitRequest, @Nullable Map<Option, ?> options)
throws SpannerException {
GrpcCallContext context = newCallContext(options, commitRequest.getSession());
return get(
getSpannerStub(SessionName.parse(commitRequest.getSession()))
.commitCallable()
.futureCall(commitRequest, context)); }
getSpannerStub(SessionName.parse(commitRequest.getSession()))
.commitCallable()
.futureCall(commitRequest, context));
}

@Override
public void rollback(RollbackRequest request, @Nullable Map<Option, ?> options)
throws SpannerException {
GrpcCallContext context = newCallContext(options, request.getSession());
get(getSpannerStub(SessionName.parse(request.getSession()))
.rollbackCallable()
.futureCall(request, context)); }
get(
getSpannerStub(SessionName.parse(request.getSession()))
.rollbackCallable()
.futureCall(request, context));
}

@Override
public PartitionResponse partitionQuery(
PartitionQueryRequest request, @Nullable Map<Option, ?> options) throws SpannerException {
GrpcCallContext context = newCallContext(options, request.getSession());
return get(
getSpannerStub(SessionName.parse(request.getSession()))
.partitionQueryCallable()
.futureCall(request, context)); }
getSpannerStub(SessionName.parse(request.getSession()))
.partitionQueryCallable()
.futureCall(request, context));
}

@Override
public PartitionResponse partitionRead(
PartitionReadRequest request, @Nullable Map<Option, ?> options) throws SpannerException {
GrpcCallContext context = newCallContext(options, request.getSession());
return get(
getSpannerStub(SessionName.parse(request.getSession()))
.partitionReadCallable()
.futureCall(request, context)); }
getSpannerStub(SessionName.parse(request.getSession()))
.partitionReadCallable()
.futureCall(request, context));
}

@Override
public Policy getDatabaseAdminIAMPolicy(String resource) {
Expand Down
12 changes: 6 additions & 6 deletions google-cloud-spanner/src/test/java/com/google/cloud/spanner/BatchCreateSessionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,12 +118,12 @@ private Spanner createSpanner(int minSessions, int maxSessions) {
.setMaxSessions(maxSessions)
.build();
SpannerOptions options =
SpannerOptions.newBuilder()
.setProjectId("[PROJECT]")
.setChannelProvider(channelProvider)
.setSessionPoolOption(sessionPoolOptions)
.setCredentials(NoCredentials.getInstance())
.build();
SpannerOptions.newBuilder()
.setProjectId("[PROJECT]")
.setChannelProvider(channelProvider)
.setSessionPoolOption(sessionPoolOptions)
.setCredentials(NoCredentials.getInstance())
.build();
return options.getService();
Comment thread
mbril marked this conversation as resolved.
Outdated
}

Expand Down
Loading