presto-main/src/main/java/io/prestosql/transaction/InMemoryTransactionManager.java/TransactionMetadata/asyncCommit
class TransactionMetadata:        public synchronized ListenableFuture<?> asyncCommit()
        {
            if (!completedSuccessfully.compareAndSet(null, true)) {
                if (completedSuccessfully.get()) {
                    // Already done
                    return immediateFuture(null);
                }
                // Transaction already aborted
                return immediateFailedFuture(new PrestoException(TRANSACTION_ALREADY_ABORTED, "Current transaction has already been aborted"));
            }

            CatalogName writeCatalogName = this.writtenConnectorId.get();
            if (writeCatalogName == null) {
                ListenableFuture<?> future = Futures.allAsList(connectorIdToMetadata.values().stream()
                        .map(transactionMetadata -> finishingExecutor.submit(transactionMetadata::commit))
                        .collect(toList()));
                addExceptionCallback(future, throwable -> {
                    abortInternal();
                    log.error(throwable, "Read-only connector should not throw exception on commit");
                });
                return nonCancellationPropagating(future);
            }

            Supplier<ListenableFuture<?>> commitReadOnlyConnectors = () -> {
                ListenableFuture<? extends List<?>> future = Futures.allAsList(connectorIdToMetadata.entrySet().stream()
                        .filter(entry -> !entry.getKey().equals(writeCatalogName))
                        .map(Entry::getValue)
                        .map(transactionMetadata -> finishingExecutor.submit(transactionMetadata::commit))
                        .collect(toList()));
                addExceptionCallback(future, throwable -> log.error(throwable, "Read-only connector should not throw exception on commit"));
                return future;
            };

            ConnectorTransactionMetadata writeConnector = connectorIdToMetadata.get(writeCatalogName);
            ListenableFuture<?> commitFuture = finishingExecutor.submit(writeConnector::commit);
            ListenableFuture<?> readOnlyCommitFuture = Futures.transformAsync(commitFuture, ignored -> commitReadOnlyConnectors.get(), directExecutor());
            addExceptionCallback(readOnlyCommitFuture, this::abortInternal);
            return nonCancellationPropagating(readOnlyCommitFuture);
        }