chore(spanner): support for TransactionManager

Author: harshachinta

google-cloud-spanner/src/main/java/com/google/cloud/spanner/DelayedTransactionManager.java

@@ -49,6 +49,11 @@ public TransactionContext begin() {
     return getTransactionManager().begin();
   }
 
+  @Override
+  public TransactionContext begin(AbortedException exception) {
+    return getTransactionManager().begin(exception);
+  }
+
   @Override
   public void commit() {
     getTransactionManager().commit();

google-cloud-spanner/src/main/java/com/google/cloud/spanner/TransactionManager.java

@@ -61,6 +61,8 @@ enum TransactionState {
    */
   TransactionContext begin();
 
+  TransactionContext begin(AbortedException exception);
+
   /**
    * Commits the currently active transaction. If the transaction was already aborted, then this
    * would throw an {@link AbortedException}.

google-cloud-spanner/src/main/java/com/google/cloud/spanner/TransactionManagerImpl.java

@@ -53,8 +53,19 @@ public void setSpan(ISpan span) {
   @Override
   public TransactionContext begin() {
     Preconditions.checkState(txn == null, "begin can only be called once");
+    return begin(ByteString.EMPTY);
+  }
+
+  @Override
+  public TransactionContext begin(AbortedException exception) {
+    Preconditions.checkState(txn == null, "begin can only be called once");
+    ByteString previousAbortedTransactionID = exception.getTransactionID() != null ? exception.getTransactionID() : ByteString.EMPTY;
+    return begin(previousAbortedTransactionID);
+  }
+
+  TransactionContext begin(ByteString previousTransactionId) {
     try (IScope s = tracer.withSpan(span)) {
-      txn = session.newTransaction(options, /* previousTransactionId = */ ByteString.EMPTY);
+      txn = session.newTransaction(options, /* previousTransactionId = */ previousTransactionId);
       session.setActive(this);
       txnState = TransactionState.STARTED;
       return txn;