@ SuppressWarnings ( "unchecked" ) private CompletableFuture < Void > completeTransaction ( TransactionId transactionId , TransactionState expectState , Function < TransactionInfo , TransactionInfo > updateFunction , Predicate < TransactionInfo > updatedPredicate , BiFunction < TransactionId , Transactional < ? > , CompletableFuture < Void > > completionFunction ) { return transactions . compute ( transactionId , ( id , info ) -> { if ( info == null ) { return null ; } else if ( info . state == expectState ) { return updateFunction . apply ( info ) ; } else { return info ; } } ) . thenCompose ( value -> { if ( value != null && updatedPredicate . test ( value . value ( ) ) ) { return Futures . allOf ( value . value ( ) . participants . stream ( ) . map ( participantInfo -> completeParticipant ( participantInfo , info -> completionFunction . apply ( transactionId , info ) ) ) ) . thenApply ( v -> null ) ; } return Futures . exceptionalFuture ( new TransactionException ( "Failed to acquire transaction lock" ) ) ; } ) ; }
Completes a transaction by modifying the transaction state to change ownership to this member and then completing the transaction based on the existing transaction state .