Example of a read write transaction. + *
{@code
+ * long singerId = my_singer_id;
+ * TransactionRunner runner = dbClient.readWriteTransaction();
+ * runner.run(
+ * new TransactionCallable() {
+ *
+ * @Override
+ * public Void run(TransactionContext transaction) throws Exception {
+ * String column = "FirstName";
+ * Struct row =
+ * transaction.readRow("Singers", Key.of(singerId), Collections.singleton(column));
+ * String name = row.getString(column);
+ * transaction.buffer(
+ * Mutation.newUpdateBuilder("Singers").set(column).to(name.toUpperCase()).build());
+ * return null;
+ * }
+ * });
+ * }
+ *
*/
TransactionRunner readWriteTransaction();
+
+ /**
+ * Returns a transaction manager which allows manual management of transaction lifecycle. This
+ * API is meant for advanced users. Most users should instead use the
+ * {@link #readWriteTransaction()} API instead.
+ *
+ * Example of using {@link TransactionManager}. + *
{@code
+ * long singerId = my_singer_id;
+ * try (TransactionManager manager = dbClient.transactionManager()) {
+ * TransactionContext txn = manager.begin();
+ * while (true) {
+ * String column = "FirstName";
+ * Struct row = txn.readRow("Singers", Key.of(singerId), Collections.singleton(column));
+ * String name = row.getString(column);
+ * txn.buffer(
+ * Mutation.newUpdateBuilder("Singers").set(column).to(name.toUpperCase()).build());
+ * try {
+ * manager.commit();
+ * break;
+ * } catch (AbortedException e) {
+ * Thread.sleep(e.getRetryDelayInMillis() / 1000);
+ * txn = manager.resetForRetry();
+ * }
+ * }
+ * }
+ * }
+ *
+ */
+ TransactionManager transactionManager();
}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java
index af7c113ca42a..b8c51849d5a6 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java
@@ -144,6 +144,17 @@ public TransactionRunner readWriteTransaction() {
}
}
+ @Override
+ public TransactionManager transactionManager() {
+ Span span = tracer.spanBuilder(READ_WRITE_TRANSACTION).startSpan();
+ try (Scope s = tracer.withSpan(span)) {
+ return pool.getReadWriteSession().transactionManager();
+ } catch (RuntimeException e) {
+ TraceUtil.endSpanWithFailure(span, e);
+ throw e;
+ }
+ }
+
ListenableFutureAt any point in time there can be at most one active transaction in this manager. When that + * transaction is committed, if it fails with an {@code ABORTED} error, calling + * {@link #resetForRetry()} would create a new {@link TransactionContext}. The newly created + * transaction would use the same session thus increasing its lock priority. If the transaction is + * committed successfully, or is rolled back or commit fails with any error other than + * {@code ABORTED}, the manager is considered complete and no further transactions are allowed to be + * created in it. + * + *
Every {@code TransactionManager} should either be committed or rolled back. Failure to do so
+ * can cause resources to be leaked and deadlocks. Easiest way to guarantee this is by calling
+ * {@link #close()} in a finally block.
+ *
+ * @see DatabaseClient#transactionManager()
+ */
+public interface TransactionManager extends AutoCloseable {
+
+ /**
+ * State of the transaction manager.
+ */
+ public enum TransactionState {
+ // Transaction has been started either by calling {@link #begin()} or via
+ // {@link resetForRetry()} but has not been commited or rolled back yet.
+ STARTED,
+ // Transaction was sucessfully committed. This is a terminal state.
+ COMMITTED,
+ // Transaction failed during commit with an error other than ABORTED. Transaction cannot be
+ // retried in this state. This is a terminal state.
+ COMMIT_FAILED,
+ // Transaction failed during commit with ABORTED and can be retried.
+ ABORTED,
+ // Transaction was rolled back. This is a terminal state.
+ ROLLED_BACK
+ }
+
+ /**
+ * Creates a new read write transaction. This must be called before doing any other operation and
+ * can only be called once. To create a new transaction for subsequent retries, see
+ * {@link #resetForRetry()}.
+ */
+ TransactionContext begin();
+
+ /**
+ * Commits the currently active transaction. If the transaction was already aborted, then this
+ * would throw an {@link AbortedException}.
+ */
+ void commit();
+
+ /**
+ * Rolls back the currently active transaction. In most cases there should be no need to call this
+ * explicitly since {@link #close()} would automatically roll back any active transaction.
+ */
+ void rollback();
+
+ /**
+ * Creates a new transaction for retry. This should only be called if the previous transaction
+ * failed with {@code ABORTED}. In all other cases, this will throw an
+ * {@link IllegalStateException}. Users should backoff before calling this method. Backoff delay
+ * is specified by {@link SpannerException#getRetryDelayInMillis()} on the
+ * {@code SpannerException} throw by the previous commit call.
+ */
+ TransactionContext resetForRetry();
+
+ /**
+ * Returns the commit timestamp if the transaction committed successfully otherwise it will throw
+ * {@code IllegalStateException}.
+ */
+ Timestamp getCommitTimestamp();
+
+ /**
+ * Returns the state of the transaction.
+ */
+ TransactionState getState();
+
+ /**
+ * Closes the manager. If there is an active transaction, it will be rolled back. Underlying
+ * session will be released back to the session pool.
+ */
+ @Override
+ void close();
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/TransactionManagerImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/TransactionManagerImpl.java
new file mode 100644
index 000000000000..7a68b8b09c64
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/TransactionManagerImpl.java
@@ -0,0 +1,133 @@
+/*
+ * Copyright 2017 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
+ *
+ * http://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 com.google.cloud.Timestamp;
+import com.google.cloud.spanner.SpannerImpl.SessionImpl;
+import com.google.cloud.spanner.SpannerImpl.SessionTransaction;
+import com.google.common.base.Preconditions;
+
+import io.opencensus.common.Scope;
+import io.opencensus.trace.Span;
+import io.opencensus.trace.Tracer;
+import io.opencensus.trace.Tracing;
+
+/**
+ * Implementation of {@link TransactionManager}.
+ */
+final class TransactionManagerImpl implements TransactionManager, SessionTransaction {
+ private static final Tracer tracer = Tracing.getTracer();
+
+ private final SessionImpl session;
+ private final Span span;
+
+ private SpannerImpl.TransactionContextImpl txn;
+ private TransactionState txnState;
+
+ TransactionManagerImpl(SessionImpl session) {
+ this.session = session;
+ this.span = Tracing.getTracer().getCurrentSpan();
+ }
+
+ @Override
+ public TransactionContext begin() {
+ Preconditions.checkState(txn == null, "begin can only be called once");
+ try (Scope s = tracer.withSpan(span)) {
+ txn = session.newTransaction();
+ session.setActive(this);
+ txn.ensureTxn();
+ txnState = TransactionState.STARTED;
+ return txn;
+ }
+ }
+
+ @Override
+ public void commit() {
+ Preconditions.checkState(txnState == TransactionState.STARTED, "commit can only be invoked if"
+ + " the transaction is in progress");
+ if (txn.isAborted()) {
+ txnState = TransactionState.ABORTED;
+ throw SpannerExceptionFactory.newSpannerException(ErrorCode.ABORTED,
+ "Transaction already aborted");
+ }
+ try {
+ txn.commit();
+ txnState = TransactionState.COMMITTED;
+ } catch (AbortedException e1) {
+ txnState = TransactionState.ABORTED;
+ throw e1;
+ } catch (SpannerException e2) {
+ txnState = TransactionState.COMMIT_FAILED;
+ throw e2;
+ }
+ }
+
+ @Override
+ public void rollback() {
+ Preconditions.checkState(txnState == TransactionState.STARTED,
+ "rollback can only be called if the transaction is in progress");
+ try {
+ txn.rollback();
+ } finally {
+ txnState = TransactionState.ROLLED_BACK;
+ }
+ }
+
+ @Override
+ public TransactionContext resetForRetry() {
+ if (txn == null
+ || !txn.isAborted() && txnState != TransactionState.ABORTED) {
+ throw new IllegalStateException("resetForRetry can only be called if the previous attempt"
+ + " aborted");
+ }
+ try (Scope s = tracer.withSpan(span)) {
+ txn = session.newTransaction();
+ txn.ensureTxn();
+ txnState = TransactionState.STARTED;
+ return txn;
+ }
+ }
+
+ @Override
+ public Timestamp getCommitTimestamp() {
+ Preconditions.checkState(txnState == TransactionState.COMMITTED,
+ "getCommitTimestamp can only be invoked if the transaction committed successfully");
+ return txn.commitTimestamp();
+ }
+
+ @Override
+ public void close() {
+ try {
+ if (txnState == TransactionState.STARTED && !txn.isAborted()) {
+ txn.rollback();
+ txnState = TransactionState.ROLLED_BACK;
+ }
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public TransactionState getState() {
+ return txnState;
+ }
+
+ @Override
+ public void invalidate() {
+ close();
+ }
+}
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/TransactionManagerImplTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/TransactionManagerImplTest.java
new file mode 100644
index 000000000000..f3f9ddf2f000
--- /dev/null
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/TransactionManagerImplTest.java
@@ -0,0 +1,161 @@
+/*
+ * Copyright 2017 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
+ *
+ * http://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 com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.MockitoAnnotations.initMocks;
+
+import com.google.cloud.spanner.SpannerImpl.SessionImpl;
+import com.google.cloud.spanner.TransactionManager.TransactionState;
+import com.google.cloud.Timestamp;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+
+@RunWith(JUnit4.class)
+public class TransactionManagerImplTest {
+
+ @Rule public ExpectedException exception = ExpectedException.none();
+
+ @Mock private SessionImpl session;
+ @Mock SpannerImpl.TransactionContextImpl txn;
+ private TransactionManagerImpl manager;
+
+ @Before
+ public void setUp() {
+ initMocks(this);
+ manager = new TransactionManagerImpl(session);
+ }
+
+ @Test
+ public void beginCalledTwiceFails() {
+ when(session.newTransaction()).thenReturn(txn);
+ assertThat(manager.begin()).isEqualTo(txn);
+ assertThat(manager.getState()).isEqualTo(TransactionState.STARTED);
+ exception.expect(IllegalStateException.class);
+ manager.begin();
+ }
+
+ @Test
+ public void commitBeforeBeginFails() {
+ exception.expect(IllegalStateException.class);
+ manager.commit();
+ }
+
+ @Test
+ public void rollbackBeforeBeginFails() {
+ exception.expect(IllegalStateException.class);
+ manager.rollback();
+ }
+
+ @Test
+ public void resetBeforeBeginFails() {
+ exception.expect(IllegalStateException.class);
+ manager.resetForRetry();
+ }
+
+ @Test
+ public void transactionRolledBackOnClose() {
+ when(session.newTransaction()).thenReturn(txn);
+ when(txn.isAborted()).thenReturn(false);
+ manager.begin();
+ manager.close();
+ verify(txn).rollback();
+ }
+
+ @Test
+ public void commitSucceeds() {
+ when(session.newTransaction()).thenReturn(txn);
+ Timestamp commitTimestamp = Timestamp.ofTimeMicroseconds(1);
+ when(txn.commitTimestamp()).thenReturn(commitTimestamp);
+ manager.begin();
+ manager.commit();
+ assertThat(manager.getState()).isEqualTo(TransactionState.COMMITTED);
+ assertThat(manager.getCommitTimestamp()).isEqualTo(commitTimestamp);
+ }
+
+ @Test
+ public void resetAfterSuccessfulCommitFails() {
+ when(session.newTransaction()).thenReturn(txn);
+ manager.begin();
+ manager.commit();
+ exception.expect(IllegalStateException.class);
+ manager.resetForRetry();
+ }
+
+ @Test
+ public void resetAfterAbortSucceeds() {
+ when(session.newTransaction()).thenReturn(txn);
+ manager.begin();
+ doThrow(SpannerExceptionFactory.newSpannerException(ErrorCode.ABORTED, ""))
+ .when(txn).commit();
+ try {
+ manager.commit();
+ fail("Expected AbortedException");
+ } catch (AbortedException e) {
+ assertThat(manager.getState()).isEqualTo(TransactionState.ABORTED);
+ }
+ txn = Mockito.mock(SpannerImpl.TransactionContextImpl.class);
+ when(session.newTransaction()).thenReturn(txn);
+ assertThat(manager.resetForRetry()).isEqualTo(txn);
+ assertThat(manager.getState()).isEqualTo(TransactionState.STARTED);
+ }
+
+ @Test
+ public void resetAfterErrorFails() {
+ when(session.newTransaction()).thenReturn(txn);
+ manager.begin();
+ doThrow(SpannerExceptionFactory.newSpannerException(ErrorCode.UNKNOWN, ""))
+ .when(txn).commit();
+ try {
+ manager.commit();
+ fail("Expected AbortedException");
+ } catch (SpannerException e) {
+ assertThat(e.getErrorCode()).isEqualTo(ErrorCode.UNKNOWN);
+ }
+ exception.expect(IllegalStateException.class);
+ manager.resetForRetry();
+ }
+
+ @Test
+ public void rollbackAfterCommitFails() {
+ when(session.newTransaction()).thenReturn(txn);
+ manager.begin();
+ manager.commit();
+ exception.expect(IllegalStateException.class);
+ manager.rollback();
+ }
+
+ @Test
+ public void commitAfterRollbackFails() {
+ when(session.newTransaction()).thenReturn(txn);
+ manager.begin();
+ manager.rollback();
+ exception.expect(IllegalStateException.class);
+ manager.commit();
+ }
+}
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/TransactionRunnerImplTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/TransactionRunnerImplTest.java
index 3818ad2dfc0d..9362814146ab 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/TransactionRunnerImplTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/TransactionRunnerImplTest.java
@@ -13,43 +13,42 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package com.google.cloud.spanner;
import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import com.google.api.client.util.BackOff;
import com.google.cloud.spanner.TransactionRunner.TransactionCallable;
import com.google.cloud.spanner.spi.v1.SpannerRpc;
-import com.google.cloud.spanner.spi.v1.SpannerRpc.Option;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.Duration;
-import com.google.protobuf.Timestamp;
-import com.google.rpc.RetryInfo;
-import com.google.spanner.v1.CommitRequest;
-import com.google.spanner.v1.CommitResponse;
import io.grpc.Context;
-import io.grpc.Metadata;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
-import io.grpc.protobuf.ProtoUtils;
-import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
-import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
-/** Unit test for {@link com.google.cloud.spanner.SpannerImpl.TransactionRunnerImpl} */
+/**
+ * Unit test for {@link com.google.cloud.spanner.SpannerImpl.TransactionRunnerImpl}
+ **/
@RunWith(JUnit4.class)
public class TransactionRunnerImplTest {
@Mock private SpannerRpc rpc;
@Mock private SpannerImpl.SessionImpl session;
@Mock private SpannerImpl.TransactionRunnerImpl.Sleeper sleeper;
+ @Mock private SpannerImpl.TransactionContextImpl txn;
private SpannerImpl.TransactionRunnerImpl transactionRunner;
private boolean firstRun;
@@ -57,89 +56,102 @@ public class TransactionRunnerImplTest {
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
firstRun = true;
+ when(session.newTransaction()).thenReturn(txn);
transactionRunner = new SpannerImpl.TransactionRunnerImpl(session, rpc, sleeper, 1);
- when(session.beginTransaction()).thenReturn(ByteString.copyFromUtf8("transaction"));
- when(session.getName()).thenReturn("fake_session");
}
@Test
- public void runAbort() {
- runTransaction(createRetryException(Status.Code.ABORTED));
- ArgumentCaptor