fix: AsyncTransactionManager did not propagate statement errors

Invalid statements or other statements that would cause an error would not cause the
returned ApiFuture to fail.

Fixes #514

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

@@ -121,6 +121,7 @@ public void onSuccess(I result) {
                       @Override
                       public void onFailure(Throwable t) {
                         mgr.onError(t);
+                        statementResult.setException(t);
                         txnResult.setException(t);
                       }
 
@@ -132,6 +133,7 @@ public void onSuccess(O result) {
                     MoreExecutors.directExecutor());
               } catch (Throwable t) {
                 mgr.onError(t);
+                statementResult.setException(t);
                 txnResult.setException(t);
               }
             }

google-cloud-spanner/src/test/java/com/google/cloud/spanner/AsyncTransactionManagerTest.java

@@ -60,6 +60,7 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -1112,4 +1113,56 @@ public ApiFuture<Void> apply(TransactionContext txn, Struct input)
       }
     }
   }
+
+  @Test
+  public void asyncTransactionManager_shouldPropagateStatementFailure()
+      throws ExecutionException, InterruptedException, TimeoutException {
+    final Statement garbledStatement =
+        Statement.newBuilder("INSERT INTO BOOKS (UUID, TITLE) VALUES ('123', 'Test book')jljlk")
+            .build();
+    mockSpanner.putStatementResult(
+        StatementResult.exception(
+            garbledStatement,
+            Status.INVALID_ARGUMENT.withDescription("Garbled SQL").asRuntimeException()));
+
+    DatabaseClient dbClient = client();
+    try (AsyncTransactionManager transactionManager = dbClient.transactionManagerAsync()) {
+      TransactionContextFuture txnContextFuture = transactionManager.beginAsync();
+      AsyncTransactionStep<Void, Long> updateFuture =
+          txnContextFuture.then(
+              new AsyncTransactionFunction<Void, Long>() {
+                @Override
+                public ApiFuture<Long> apply(TransactionContext txn, Void input) throws Exception {
+                  return txn.executeUpdateAsync(garbledStatement);
+                }
+              },
+              executor);
+      final SettableApiFuture<Void> res = SettableApiFuture.create();
+      ApiFutures.addCallback(
+          updateFuture,
+          new ApiFutureCallback<Long>() {
+            @Override
+            public void onFailure(Throwable throwable) {
+              // Check that we got the expected failure.
+              try {
+                assertThat(throwable).isInstanceOf(SpannerException.class);
+                SpannerException e = (SpannerException) throwable;
+                assertThat(e.getErrorCode()).isEqualTo(ErrorCode.INVALID_ARGUMENT);
+                assertThat(e.getMessage()).contains("Garbled SQL");
+                res.set(null);
+              } catch (Throwable t) {
+                res.setException(t);
+              }
+            }
+
+            @Override
+            public void onSuccess(Long aLong) {
+              res.setException(new AssertionError("Statement should not succeed."));
+            }
+          },
+          executor);
+
+      assertThat(res.get(10L, TimeUnit.SECONDS)).isNull();
+    }
+  }
 }