Transaction Support for mongoDB with Reactive flow

[doogle-oss]

I have been trying out various options on using transactions with reactive flow on Quarkus 3.0.
I have made some headway and simple ones work just fine but need to know the right way to implement the same
@transaction does not seem to work when there is failures and mongo db documents update.

• quarkus reproducer repo
  What i tried till now

Mapper for reactivestreams publisher to concurrent flow publisher

  private Uni<Void> abortTransaction(ClientSession sess) {
    return Uni.createFrom().publisher(AdaptersToFlow.publisher(sess.abortTransaction()))
        .log("Transaction Aborted");
  }

  private Uni<Void> commitTransaction(ClientSession sess) {
    return Uni.createFrom().publisher(AdaptersToFlow.publisher(sess.commitTransaction()))
        .log("Transaction Committed");
  }

Actual Logic for multi doc transaction in reactive fashion

  @Mutation("accountTransfer")
  public Uni<List<AccountModel>> accountTransfer(String fromAccountIdentifier,
      String toAccountIdentifier, Double amountToTransfer) {
    Log.infov("transferring {0} Hypnotons from {1} to {2}", amountToTransfer, fromAccountIdentifier,
        toAccountIdentifier);
    return client.startSession()
        .flatMap(s -> {
          Log.infov("session active transaction {0}", s.hasActiveTransaction());
          s.startTransaction();
          Log.infov("session {0}", s);
          Uni<AccountEntity> sourceAccount =
              AccountEntity.getCollection()
                  .findOneAndUpdate(s, new Document("accountIdentifier", fromAccountIdentifier),
                      inc("accountBalance", amountToTransfer * (-1)),
                      new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER)
                          .upsert(true))
                  .log("SOURCE_ACCOUNT_AFTER");
          Uni<AccountEntity> targetAccount =
              AccountEntity.getCollection()
                  .findOneAndUpdate(s, new Document("accountIdentifier", toAccountIdentifier),
                      inc("accountBalance", amountToTransfer),
                      new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER)
                          .upsert(true)).onFailure().retry()
                  .withBackOff(Duration.ofMillis(100), Duration.ofSeconds(1)).atMost(5)
                  .log("TARGET_ACCOUNT_AFTER");
          Log.infov("Is there a active transaction {0}", s.hasActiveTransaction());
          return Uni.combine().all().unis(sourceAccount, targetAccount).asTuple().log("ACCOUNTS")
              .invoke(this::checkTransaction).map(t -> List.of(t.getItem1(), t.getItem2()))
              .map(acc -> accountMapper.fromAccountEntities(acc))
              .call(v -> commitTransaction(s)).onFailure().call(
                  error -> abortTransaction(s).onFailure().retry()
                      .withBackOff(Duration.ofSeconds(2), Duration.ofSeconds(5)).atMost(5)
                      .eventually(s::close));
        });
  }

  private void checkTransaction(Tuple2<AccountEntity, AccountEntity> t) {
    if (t.getItem1().getAccountBalance() < 0 || t.getItem2().getAccountBalance() < 0) {
      throw new UnsupportedOperationException("Funds Insufficient");
    }
  }

not sure why but i had to add the following for the second transaction

.onFailure().retry()
                  .withBackOff(Duration.ofMillis(100), Duration.ofSeconds(1)).atMost(5)
                  .log("TARGET_ACCOUNT_AFTER");

References:
[1] how-to-efficiently-handle-transactions-with-the-reactive-java-driver-clarifications-on-api
[2] mongodb-multi-document-acid-transactions-general-availability
[3] [MongoDB Transactions Docs] (https://www.mongodb.com/docs/manual/core/transactions)
[4] [Java MongoDB Transactions] (https://hevodata.com/learn/java-mongodb-transactions/)
[5] [how-to-create-transactions-using-mongodb-java] (https://stackoverflow.com/questions/63997699/how-to-create-transactions-using-mongodb-java)
[6] [Open Jira Epic for OOB Support] (https://jira.mongodb.org/browse/JAVA-3222)
[7] [Actual Jira with mongodb - Implement convenient reactive API for Transactions] (https://jira.mongodb.org/browse/JAVA-3539)
[8] More ref
[9] [More reference] (https://github.com/mongodb/mongo-java-driver-reactivestreams/blob/master/driver/src/main/com/mongodb/reactivestreams/client/ClientSession.java)
https://www.mongodb.com/community/forums/t/given-transaction-number-does-not-match-any-in-progress-transactions/703/4

https://stackoverflow.com/questions/58487982/how-to-prevent-given-transaction-number-1-does-not-match-any-in-progress-transa

https://www.mongodb.com/community/forums/t/given-transaction-number-does-not-match-any-in-progress-transactions-with-mongo-java/223755
https://www.digitalocean.com/community/tutorials/how-to-use-transactions-in-mongodb

[doogle-oss]

@loicmathieu @cescoffier If you can direct in some way it will help do it the right way.

[loicmathieu]

@Debu999 this functionality will be provided by Quarkus 3.2, see the PR #32794

[doogle-oss]

tested and validated. Thx @loicmathieu

[doogle-oss]

Closing the same as its already made available from 3.2 in quarkus.

[doogle-oss]

@Mutation("accountTransfer")
  public Uni<List<AccountModel>> accountTransfer(String fromAccountIdentifier,
      String toAccountIdentifier, Double amountToTransfer) {
    Log.infov("transferring {0} Hypnotons from {1} to {2}", amountToTransfer, fromAccountIdentifier,
        toAccountIdentifier);

    return withTransaction(() -> {
      Uni<AccountEntity> sourceAccount = AccountEntity.getCollection()
          .findOneAndUpdate(new Document("accountIdentifier", fromAccountIdentifier),
              inc("accountBalance", amountToTransfer * (-1)),
              new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER).upsert(true))
          .log("SOURCE_ACCOUNT_AFTER");

      Uni<AccountEntity> targetAccount = AccountEntity.getCollection()
          .findOneAndUpdate(new Document("accountIdentifier", toAccountIdentifier),
              inc("accountBalance", amountToTransfer),
              new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER).upsert(true))
//                      .onFailure().retry()
//                  .withBackOff(Duration.ofMillis(100), Duration.ofSeconds(1)).atMost(5)
          .log("TARGET_ACCOUNT_AFTER");
      return Uni.combine().all().unis(sourceAccount, targetAccount).asTuple().log("ACCOUNTS")
          .invoke(this::checkTransaction).map(t -> List.of(t.getItem1(), t.getItem2()))
          .map(acc -> accountMapper.fromAccountEntities(acc));
    });
  }

  private void checkTransaction(Tuple2<AccountEntity, AccountEntity> t) {
    if (t.getItem1().getAccountBalance() < 0 || t.getItem2().getAccountBalance() < 0) {
      throw new UnsupportedOperationException("Funds Insufficient");
    }
  }

[doogle-oss]

works like a charm @loicmathieu Thanks for bringing the feature in.