Exceptions not propagated from async function back to async transaction

[elefeint]

Environment details

1. Java version: 11
2. spanner version(s): 1.59.0

Steps to reproduce

In the Async API, TransactionContext.executeUpdateAsync() correctly propagates errors through ApiFuture callbacks. However, in an asynchronous transaction, the .then() operator does not get the same exceptions propagated to it from the transaction function.

Code example

The following test (just add a custom databaseId) hangs because neither the success nor failure callback is called, and the latch is never released. If the SQL statement were corrected, the test would pass, as success is propagated correctly.

@Test
  public void clientLibraryTransactionNotPropagatingFailure()
      throws ExecutionException, InterruptedException {
    Executor executor = Executors.newFixedThreadPool(4);

    SpannerOptions options = SpannerOptions.newBuilder().build();
    DatabaseClient dbClient = options.getService().getDatabaseClient(databaseId);
    AsyncTransactionManager transactionManager = dbClient.transactionManagerAsync();
    TransactionContextFuture txnContextFuture = transactionManager.beginAsync();


    // Statement has garbled SQL syntax intentionally
    Statement statement =
        Statement.newBuilder("INSERT INTO BOOKS (UUID, TITLE) VALUES ('123', 'Test book')jljlk")
            .build();
    ApiFuture<Long> updateFuture =
        txnContextFuture.then(
            (txnContext, unusedVoid) -> txnContext.executeUpdateAsync(statement),
            executor);

    CountDownLatch latch = new CountDownLatch(1);
    ApiFutures.addCallback(
        updateFuture,
        new ApiFutureCallback<Long>() {
          @Override
          public void onFailure(Throwable throwable) {
            // Expecting the failure callback to be called, but it's not, so the latch hangs.
            System.out.println("Exception: " + throwable);
            latch.countDown();
          }

          @Override
          public void onSuccess(Long aLong) {
            System.out.println("Success: " + aLong);
            latch.countDown();
          }
        }, executor);

    latch.await();
  }

Thanks!