+ final ListenableFuture<T> ret = broker.commit(transaction, cohorts, futureValueSupplier);
+
+ COUNTER_UPDATER.incrementAndGet(this);
+ Futures.addCallback(ret, new FutureCallback<T>() {
+ @Override
+ public void onSuccess(final T result) {
+ transactionCompleted();
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ transactionFailed(transaction, throwable);
+ }
+ }, MoreExecutors.directExecutor());
+
+ return ret;