CompletableFuture<TxnStatus> sealTxnBody(final String host,
                                             final String scope,
                                             final String stream,
                                             final boolean commit,
                                             final UUID txnId,
                                             final Version version,
                                             final OperationContext ctx) {
        TxnResource resource = new TxnResource(scope, stream, txnId);
        Optional<Version> versionOpt = Optional.ofNullable(version);

        // Step 1. Add txn to current host's index, if it is not already present
        CompletableFuture<Void> addIndex = host.equals(hostId) && !timeoutService.containsTxn(scope, stream, txnId) ?
                // PS: txn version in index does not matter, because if update is successful,
                // then txn would no longer be open.
                streamMetadataStore.addTxnToIndex(hostId, resource, version) :
                CompletableFuture.completedFuture(null);

        addIndex.whenComplete((v, e) -> {
            if (e != null) {
                log.debug("Txn={}, already present/newly added to host-txn index of host={}", txnId, hostId);
            } else {
                log.debug("Txn={}, added txn to host-txn index of host={}", txnId, hostId);
            }
        });

        // Step 2. Seal txn
        CompletableFuture<AbstractMap.SimpleEntry<TxnStatus, Integer>> sealFuture = addIndex.thenComposeAsync(x ->
                streamMetadataStore.sealTransaction(scope, stream, txnId, commit, versionOpt, ctx, executor), executor)
                .whenComplete((v, e) -> {
                    if (e != null) {
                        log.debug("Txn={}, failed sealing txn", txnId);
                    } else {
                        log.debug("Txn={}, sealed successfully, commit={}", txnId, commit);
                    }
                });

        // Step 3. write event to corresponding stream.
        return sealFuture.thenComposeAsync(pair -> {
            TxnStatus status = pair.getKey();
            switch (status) {
                case COMMITTING:
                    return writeCommitEvent(scope, stream, pair.getValue(), txnId, status);
                case ABORTING:
                    return writeAbortEvent(scope, stream, pair.getValue(), txnId, status);
                case ABORTED:
                case COMMITTED:
                    return CompletableFuture.completedFuture(status);
                case OPEN:
                case UNKNOWN:
                default:
                    // Not possible after successful streamStore.sealTransaction call, because otherwise an
                    // exception would be thrown.
                    return CompletableFuture.completedFuture(status);
            }
        }, executor).thenComposeAsync(status -> {
            // Step 4. Remove txn from timeoutService, and from the index.
            timeoutService.removeTxn(scope, stream, txnId);
            log.debug("Txn={}, removed from timeout service", txnId);
            return streamMetadataStore.removeTxnFromIndex(host, resource, true).whenComplete((v, e) -> {
                if (e != null) {
                    log.debug("Txn={}, failed removing txn from host-txn index of host={}", txnId, hostId);
                } else {
                    log.debug("Txn={}, removed txn from host-txn index of host={}", txnId, hostId);
                }
            }).thenApply(x -> status);
        }, executor);
    }
CompletableFuture < TxnStatus > sealTxnBody ( final String host , final String scope , final String stream , final boolean commit , final UUID txnId , final Version version , final OperationContext ctx ) { TxnResource resource = new TxnResource ( scope , stream , txnId ) ; Optional < Version > versionOpt = Optional . ofNullable ( version ) ; CompletableFuture < Void > addIndex = host . equals ( hostId ) && ! timeoutService . containsTxn ( scope , stream , txnId ) ? streamMetadataStore . addTxnToIndex ( hostId , resource , version ) : CompletableFuture . completedFuture ( null ) ; addIndex . whenComplete ( ( v , e ) -> { if ( e != null ) { log . debug ( "Txn={}, already present/newly added to host-txn index of host={}" , txnId , hostId ) ; } else { log . debug ( "Txn={}, added txn to host-txn index of host={}" , txnId , hostId ) ; } } ) ; CompletableFuture < AbstractMap . SimpleEntry < TxnStatus , Integer > > sealFuture = addIndex . thenComposeAsync ( x -> streamMetadataStore . sealTransaction ( scope , stream , txnId , commit , versionOpt , ctx , executor ) , executor ) . whenComplete ( ( v , e ) -> { if ( e != null ) { log . debug ( "Txn={}, failed sealing txn" , txnId ) ; } else { log . debug ( "Txn={}, sealed successfully, commit={}" , txnId , commit ) ; } } ) ; return sealFuture . thenComposeAsync ( pair -> { TxnStatus status = pair . getKey ( ) ; switch ( status ) { case COMMITTING : return writeCommitEvent ( scope , stream , pair . getValue ( ) , txnId , status ) ; case ABORTING : return writeAbortEvent ( scope , stream , pair . getValue ( ) , txnId , status ) ; case ABORTED : case COMMITTED : return CompletableFuture . completedFuture ( status ) ; case OPEN : case UNKNOWN : default : return CompletableFuture . completedFuture ( status ) ; } } , executor ) . thenComposeAsync ( status -> { timeoutService . removeTxn ( scope , stream , txnId ) ; log . debug ( "Txn={}, removed from timeout service" , txnId ) ; return streamMetadataStore . removeTxnFromIndex ( host , resource , true ) . whenComplete ( ( v , e ) -> { if ( e != null ) { log . debug ( "Txn={}, failed removing txn from host-txn index of host={}" , txnId , hostId ) ; } else { log . debug ( "Txn={}, removed txn from host-txn index of host={}" , txnId , hostId ) ; } } ) . thenApply ( x -> status ) ; } , executor ) ; }
Seals a txn and transitions it to COMMITTING (resp. ABORTING) state if commit param is true (resp. false).

Post-condition:
1. If seal completes successfully, then
(a) txn state is COMMITTING/ABORTING,
(b) CommitEvent/AbortEvent is present in the commit stream/abort stream,
(c) txn is removed from host-txn index,
(d) txn is removed from the timeout service.

2. If process fails after transitioning txn to COMMITTING/ABORTING state, but before responding to client, then
since txn is present in the host-txn index, some other controller process shall put CommitEvent/AbortEvent to
commit stream/abort stream.

@param host    host id. It is different from hostId iff invoked from TxnSweeper for aborting orphaned txn.
@param scope   scope name.
@param stream  stream name.
@param commit  boolean indicating whether to commit txn.
@param txnId   txn id.
@param version expected version of txn node in store.
@param ctx     context.
@return        Txn status after sealing it.