Skip to content

Commit 4f31c44

Browse files
vkediavkedia
authored
---
yaml --- r: 9297 b: refs/heads/master c: c0d4b28 h: refs/heads/master i: 9295: 2db9762
1 parent af172cf commit 4f31c44

11 files changed

Lines changed: 796 additions & 79 deletions

File tree

‎[refs]‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
---
2-
refs/heads/master: 7d88150fd887e9b4a1a4820545fc2bfc7996c70e
2+
refs/heads/master: c0d4b28dddbcbfca3456c88f99d2d936a9eb1ca8
33
refs/heads/travis: 47e4fee4fd5af9b2a8ce46f23c72ec95f9b195b2
44
refs/heads/gh-pages: 1e525ea89a8c2133b947a8f1e3306079d006152d
55
refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444

‎trunk/google-cloud-examples/src/main/java/com/google/cloud/examples/spanner/snippets/DatabaseClientSnippets.java‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,15 @@
2323
package com.google.cloud.examples.spanner.snippets;
2424

2525
import com.google.cloud.Timestamp;
26+
import com.google.cloud.spanner.AbortedException;
2627
import com.google.cloud.spanner.DatabaseClient;
2728
import com.google.cloud.spanner.Key;
2829
import com.google.cloud.spanner.Mutation;
2930
import com.google.cloud.spanner.ReadOnlyTransaction;
3031
import com.google.cloud.spanner.Struct;
3132
import com.google.cloud.spanner.TimestampBound;
3233
import com.google.cloud.spanner.TransactionContext;
34+
import com.google.cloud.spanner.TransactionManager;
3335
import com.google.cloud.spanner.TransactionRunner;
3436
import com.google.cloud.spanner.TransactionRunner.TransactionCallable;
3537
import java.util.Collections;
@@ -222,4 +224,31 @@ public Void run(TransactionContext transaction) throws Exception {
222224
});
223225
// [END readWriteTransaction]
224226
}
227+
228+
/**
229+
* Example of using {@link TransactionManager}.
230+
*/
231+
// [TARGET transactionManager()]
232+
// [VARIABLE my_singer_id]
233+
public void transactionManager(final long singerId) throws InterruptedException {
234+
// [START transactionManager]
235+
try (TransactionManager manager = dbClient.transactionManager()) {
236+
TransactionContext txn = manager.begin();
237+
while (true) {
238+
String column = "FirstName";
239+
Struct row = txn.readRow("Singers", Key.of(singerId), Collections.singleton(column));
240+
String name = row.getString(column);
241+
txn.buffer(
242+
Mutation.newUpdateBuilder("Singers").set(column).to(name.toUpperCase()).build());
243+
try {
244+
manager.commit();
245+
break;
246+
} catch (AbortedException e) {
247+
Thread.sleep(e.getRetryDelayInMillis() / 1000);
248+
txn = manager.resetForRetry();
249+
}
250+
}
251+
}
252+
// [END transactionManager]
253+
}
225254
}

‎trunk/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClient.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -237,6 +237,56 @@ public interface DatabaseClient {
237237
* });
238238
* </code></pre>
239239
*
240+
* <p>Example of a read write transaction.
241+
* <pre> {@code
242+
* long singerId = my_singer_id;
243+
* TransactionRunner runner = dbClient.readWriteTransaction();
244+
* runner.run(
245+
* new TransactionCallable<Void>() {
246+
*
247+
* @Override
248+
* public Void run(TransactionContext transaction) throws Exception {
249+
* String column = "FirstName";
250+
* Struct row =
251+
* transaction.readRow("Singers", Key.of(singerId), Collections.singleton(column));
252+
* String name = row.getString(column);
253+
* transaction.buffer(
254+
* Mutation.newUpdateBuilder("Singers").set(column).to(name.toUpperCase()).build());
255+
* return null;
256+
* }
257+
* });
258+
* }</pre>
259+
*
240260
*/
241261
TransactionRunner readWriteTransaction();
262+
263+
/**
264+
* Returns a transaction manager which allows manual management of transaction lifecycle. This
265+
* API is meant for advanced users. Most users should instead use the
266+
* {@link #readWriteTransaction()} API instead.
267+
*
268+
* <p>Example of using {@link TransactionManager}.
269+
* <pre> {@code
270+
* long singerId = my_singer_id;
271+
* try (TransactionManager manager = dbClient.transactionManager()) {
272+
* TransactionContext txn = manager.begin();
273+
* while (true) {
274+
* String column = "FirstName";
275+
* Struct row = txn.readRow("Singers", Key.of(singerId), Collections.singleton(column));
276+
* String name = row.getString(column);
277+
* txn.buffer(
278+
* Mutation.newUpdateBuilder("Singers").set(column).to(name.toUpperCase()).build());
279+
* try {
280+
* manager.commit();
281+
* break;
282+
* } catch (AbortedException e) {
283+
* Thread.sleep(e.getRetryDelayInMillis() / 1000);
284+
* txn = manager.resetForRetry();
285+
* }
286+
* }
287+
* }
288+
* }</pre>
289+
*
290+
*/
291+
TransactionManager transactionManager();
242292
}

‎trunk/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,17 @@ public TransactionRunner readWriteTransaction() {
144144
}
145145
}
146146

147+
@Override
148+
public TransactionManager transactionManager() {
149+
Span span = tracer.spanBuilder(READ_WRITE_TRANSACTION).startSpan();
150+
try (Scope s = tracer.withSpan(span)) {
151+
return pool.getReadWriteSession().transactionManager();
152+
} catch (RuntimeException e) {
153+
TraceUtil.endSpanWithFailure(span, e);
154+
throw e;
155+
}
156+
}
157+
147158
ListenableFuture<Void> closeAsync() {
148159
return pool.closeAsync();
149160
}

‎trunk/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionPool.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,6 +195,65 @@ public Timestamp getReadTimestamp() {
195195
return txn.getReadTimestamp();
196196
}
197197
}
198+
199+
private static class AutoClosingTransactionManager implements TransactionManager {
200+
final TransactionManager delegate;
201+
final PooledSession session;
202+
203+
AutoClosingTransactionManager(TransactionManager delegate, PooledSession session) {
204+
this.delegate = delegate;
205+
this.session = session;
206+
}
207+
208+
@Override
209+
public TransactionContext begin() {
210+
return delegate.begin();
211+
}
212+
213+
@Override
214+
public void commit() {
215+
try {
216+
delegate.commit();
217+
} finally {
218+
if (getState() != TransactionState.ABORTED) {
219+
close();
220+
}
221+
}
222+
}
223+
224+
@Override
225+
public void rollback() {
226+
try {
227+
delegate.rollback();
228+
} finally {
229+
close();
230+
}
231+
}
232+
233+
@Override
234+
public TransactionContext resetForRetry() {
235+
return delegate.resetForRetry();
236+
}
237+
238+
@Override
239+
public Timestamp getCommitTimestamp() {
240+
return delegate.getCommitTimestamp();
241+
}
242+
243+
@Override
244+
public void close() {
245+
try {
246+
delegate.close();
247+
} finally {
248+
session.close();
249+
}
250+
}
251+
252+
@Override
253+
public TransactionState getState() {
254+
return delegate.getState();
255+
}
256+
}
198257

199258
// Exception class used just to track the stack trace at the point when a session was handed out
200259
// from the pool.
@@ -386,6 +445,12 @@ private void keepAlive() {
386445
private void markUsed() {
387446
lastUseTime = clock.instant();
388447
}
448+
449+
@Override
450+
public TransactionManager transactionManager() {
451+
markUsed();
452+
return new AutoClosingTransactionManager(delegate.transactionManager(), this);
453+
}
389454
}
390455

391456
private static final class SessionOrError {

‎trunk/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerImpl.java‎

Lines changed: 20 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -898,14 +898,25 @@ public Transaction call() throws Exception {
898898
}
899899
}
900900

901-
private <T extends SessionTransaction> T setActive(@Nullable T ctx) {
901+
TransactionContextImpl newTransaction() {
902+
TransactionContextImpl txn = new TransactionContextImpl(this, readyTransactionId, rpc,
903+
defaultPrefetchChunks);
904+
return txn;
905+
}
906+
907+
<T extends SessionTransaction> T setActive(@Nullable T ctx) {
902908
if (activeTransaction != null) {
903909
activeTransaction.invalidate();
904910
}
905911
activeTransaction = ctx;
906912
readyTransactionId = null;
907913
return ctx;
908914
}
915+
916+
@Override
917+
public TransactionManager transactionManager() {
918+
return new TransactionManagerImpl(this);
919+
}
909920
}
910921

911922
/**
@@ -914,7 +925,7 @@ private <T extends SessionTransaction> T setActive(@Nullable T ctx) {
914925
* transactions, and read-write transactions. The defining characteristic is that a session may
915926
* only have one such transaction active at a time.
916927
*/
917-
private interface SessionTransaction {
928+
static interface SessionTransaction {
918929
/** Invalidates the transaction, generally because a new one has been started on the session. */
919930
void invalidate();
920931
}
@@ -1018,7 +1029,7 @@ ResultSet executeQueryInternalWithOptions(
10181029
ExecuteSqlRequest.newBuilder()
10191030
.setSql(statement.getSql())
10201031
.setQueryMode(queryMode)
1021-
.setSession(session.name);
1032+
.setSession(session.getName());
10221033
Map<String, Value> stmtParameters = statement.getParameters();
10231034
if (!stmtParameters.isEmpty()) {
10241035
com.google.protobuf.Struct.Builder paramsBuilder = builder.getParamsBuilder();
@@ -1217,10 +1228,7 @@ void backoffSleep(Context context, long backoffMillis) {
12171228
this.session = session;
12181229
this.sleeper = sleeper;
12191230
this.span = Tracing.getTracer().getCurrentSpan();
1220-
ByteString transactionId = session.readyTransactionId;
1221-
session.readyTransactionId = null;
1222-
this.txn = new TransactionContextImpl(session, transactionId, rpc, defaultPrefetchChunks,
1223-
span);
1231+
this.txn = session.newTransaction();
12241232
}
12251233

12261234
TransactionRunnerImpl(SessionImpl session, SpannerRpc rpc, int defaultPrefetchChunks) {
@@ -1230,7 +1238,7 @@ void backoffSleep(Context context, long backoffMillis) {
12301238
@Nullable
12311239
@Override
12321240
public <T> T run(TransactionCallable<T> callable) {
1233-
try {
1241+
try (Scope s = tracer.withSpan(span)) {
12341242
return runInternal(callable);
12351243
} catch (RuntimeException e) {
12361244
TraceUtil.endSpanWithFailure(span, e);
@@ -1244,6 +1252,7 @@ private <T> T runInternal(TransactionCallable<T> callable) {
12441252
BackOff backoff = newBackOff();
12451253
final Context context = Context.current();
12461254
int attempt = 0;
1255+
// TODO: Change this to use TransactionManager.
12471256
while (true) {
12481257
checkState(
12491258
isValid, "TransactionRunner has been invalidated by a new operation on the session");
@@ -1318,7 +1327,7 @@ public void invalidate() {
13181327

13191328
private void backoff(Context context, BackOff backoff) {
13201329
long delay = txn.getRetryDelayInMillis(backoff);
1321-
txn = new TransactionContextImpl(session, null, txn.rpc, txn.defaultPrefetchChunks, span);
1330+
txn = session.newTransaction();
13221331
span.addAnnotation("Backing off",
13231332
ImmutableMap.of("Delay", AttributeValue.longAttributeValue(delay)));
13241333
sleeper.backoffSleep(context, delay);
@@ -1344,9 +1353,8 @@ static class TransactionContextImpl extends AbstractReadContext implements Trans
13441353
SessionImpl session,
13451354
@Nullable ByteString transactionId,
13461355
SpannerRpc rpc,
1347-
int defaultPrefetchChunks,
1348-
Span span) {
1349-
super(session, rpc, defaultPrefetchChunks, span);
1356+
int defaultPrefetchChunks) {
1357+
super(session, rpc, defaultPrefetchChunks);
13501358
this.transactionId = transactionId;
13511359
}
13521360

0 commit comments

Comments
 (0)