fix: handle thrown exceptions in runAsyncTransaction callback

google-cloud-firestore/src/main/java/com/google/cloud/firestore/TransactionRunner.java

@@ -146,24 +146,37 @@ public void run() {
    */
   private SettableApiFuture<T> invokeUserCallback() {
     final SettableApiFuture<T> callbackResult = SettableApiFuture.create();
+
     userCallbackExecutor.execute(
         new Runnable() {
           @Override
           public void run() {
-            ApiFutures.addCallback(
-                userCallback.updateCallback(transaction),
-                new ApiFutureCallback<T>() {
-                  @Override
-                  public void onFailure(Throwable t) {
-                    callbackResult.setException(t);
-                  }
+            try {
+              ApiFutures.addCallback(
+                  ApiFutures.catchingAsync(
+                      userCallback.updateCallback(transaction),
+                      Throwable.class,
+                      new ApiAsyncFunction<Throwable, T>() {
+                        public ApiFuture<T> apply(Throwable throwable) throws Exception {
+                          return ApiFutures.immediateFailedFuture(throwable);
+                        }
+                      },
+                      MoreExecutors.directExecutor()),
+                  new ApiFutureCallback<T>() {
+                    @Override
+                    public void onFailure(Throwable t) {
+                      callbackResult.setException(t);
+                    }
 
-                  @Override
-                  public void onSuccess(T result) {
-                    callbackResult.set(result);
-                  }
-                },
-                MoreExecutors.directExecutor());
+                    @Override
+                    public void onSuccess(T result) {
+                      callbackResult.set(result);
+                    }
+                  },
+                  firestoreExecutor);
+            } catch (Exception e) {
+              callbackResult.setException(e);
+            }
           }
         });
     return callbackResult;

google-cloud-firestore/src/test/java/com/google/cloud/firestore/TransactionTest.java

@@ -253,7 +253,7 @@ public String updateCallback(Transaction transaction) throws Exception {
   }
 
   @Test
-  public void rollbackOnCallbackErrorAsync() {
+  public void rollbackOnCallbackApiFutureErrorAsync() {
     doReturn(beginResponse())
         .doReturn(rollbackResponse())
         .when(firestoreMock)
@@ -339,6 +339,31 @@ public ApiFuture<String> updateCallback(Transaction transaction) {
     assertEquals(1, requests.size());
   }
 
+  @Test
+  public void noRollbackOnThrownExceptionAsync() {
+    doReturn(beginResponse())
+        .doReturn(rollbackResponse())
+        .when(firestoreMock)
+        .sendRequest(requestCapture.capture(), Matchers.<UnaryCallable<Message, Message>>any());
+
+    ApiFuture<String> transaction =
+        firestoreMock.runAsyncTransaction(
+            new Transaction.AsyncFunction<String>() {
+              @Override
+              public ApiFuture<String> updateCallback(Transaction transaction) {
+                throw new RuntimeException("User exception");
+              }
+            },
+            options);
+
+    try {
+      transaction.get();
+      fail();
+    } catch (Exception e) {
+      assertTrue(e.getMessage().endsWith("User exception"));
+    }
+  }
+
   @Test
   public void limitsRetriesWithFailure() {
     ResponseStubber responseStubber =

google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITSystemTest.java

@@ -688,6 +688,24 @@ public String updateCallback(Transaction transaction) {
     }
   }
 
+  @Test
+  public void asyncTxFailsWithUserError() throws Exception {
+    try {
+      firestore
+          .runAsyncTransaction(
+              new Transaction.AsyncFunction<String>() {
+                @Override
+                public ApiFuture<String> updateCallback(Transaction transaction) {
+                  throw new RuntimeException("User exception");
+                }
+              })
+          .get();
+      fail();
+    } catch (Exception e) {
+      assertTrue(e.getMessage().endsWith("User exception"));
+    }
+  }
+
   @Test
   public void doesNotRetryTransactionsWithFailedPreconditions() {
     final DocumentReference documentReference = randomColl.document();