Concurrent execution inconsistencies when using transformToUniAndMerge

[nomans1]

I have an application using Quarkus 2.15.0 and its primary purpose is to consume a Kafka event and then call Rest clients to populate information in Mongo. My requirement is to call three Rest clients, let's call them RestA, RestB and RestC.

The Kafka message contains ids which are streamed using MultiEmitter like this:

    @Incoming("event-in")
    public Uni<Void> consumeEvent(Record<String, EventMessage> eventMessage) {

        return Multi.createFrom().emitter(multiEmitter -> {
                    getParty(eventMessage.value().getUserInfo())
                            .stream()
                            .filter(Objects::nonNull)
                            .forEach(multiEmitter::emit);
                    multiEmitter.complete();
                })
                .onItem()
                .transformToUniAndMerge(party -> processParties(id, (Party) party, anotherId))
                .toUni();

processParties method calls RestA, updates the database with the information and then calls RestB and RestC concurrently. The Rest calls depends on the number of ids, so if there are 10 then the Rest clients are called 10 times. The code is kinda like this

    private Uni<Void> processParties(String id, Party party, String anotherId) {
        return callRestA(id,party)
                .chain(response -> {
                    dbService.updateAccounts(id, party, party.getAccounts());
                    List<Account> accounts = Optional.of(party)
                            .map(Party::getAccounts)
                            .orElse(List.of(new Account()));

                    return Multi.createFrom().emitter(accountsEmitter -> {
                        accounts.stream().filter(Objects::nonNull).forEach(accountsEmitter::emit);
                        accountsEmitter.complete();
                    })
                            .onItem()
                            .transformToUniAndMerge(account -> processRestBAndRestC(id, anotherId, (Account) account))
                            .toUni()
                            .invoke(() -> LOGGER.info(" processing complete for: {}", id));

                });
    }

The processRestBAndRestC method looks kinda like this. It calls RestB and RestC and updates the db with the info received. I'm using Uni.join so I can execute them concurrently and since they both have the same method signature.

  Uni<Boolean> processRestB = callRestB(id, account, anotherId).chain(res -> Uni.createFrom().item(dbService.updateRestB(id, account, anotherId, res)))
  Uni<Boolean> processRestC = callRestC(id, account, anotherId).chain(res -> Uni.createFrom().item(dbService.updateRestC(id, account, anotherId, res)))

  return Uni.join().all(processRestB, processRestC).andCollectFailures().replaceWithVoid();

I hope this rudimentary code highlights what I'm trying to do, basically my issue comes when I call transformToUniAndMerge(account -> processRestBAndRestC(id, anotherId, (Account) account)). It sometimes doesn't update the db with the info and sometimes does. If I switch to using transformToUniAndConcatenate it works perfectly fine but its slower cause it waits for the Uni to complete before moving onto the next item and I don't necessarily care about preserving the order. I just want it to consistently update the Db with all the info no matter the order I receive the data in.

Is there any way I can use the merge and get rid off the inconsistencies in my db update operations? Maybe I'm doing something wrong so would appreciate any guidance

[jponge]

Hi, Quarkus 2.15.0 ships with an old version of Mutiny.

1. How does the code behave on 3.6.6?
2. Can you extract a reproducer that's not using any database driver / Kafka

Thanks!

[nomans1]

Hey @jponge - we've built some internal libs for Quarkus and they're on Quarkus 2. I'm in the process of upgrading them so that's the reason for sticking to 2.15.0. Not sure if I can force Mutiny 3.6.6?

By reproducer you mean just use generic objects and logic and remove any db logic?

I guess my main question that I'm struggling with is, if I want to use transformtoUniAndMerge, how can I ensure that for each item it asynchronously updates the item result in a db. While using transformToUniAndConcatenate() on the original multi it the order but does not call the transforming unis until the previous one completes. Makes total time consumed increase by the number of items

[jponge]

By reproducer you mean just use generic objects and logic and remove any db logic?

Yes, it would be better to reason in pure Mutiny APIs and not involve databases and stuff

[nomans1]

Does the below work? I've cleaned it up somewhat and removed any db calls. Assume they're methods just updating or mapping objects

    public Uni<Void> consumeEvent(EventMessage eventMessage) {
        final String tokenId = eventMessage.key();
        final String userId = eventMessage.value().getCustomer().getUserId();
        /*
         * The getCustomer() method contains the below information
         * userId
         * userInfo object -> name, address, entityType, id etc
         */

         /*
          * The Party object contains -> role, customerId, type etc
          */
        return Multi.createFrom().emitter(multiEmitter -> {
                    getParty(eventMessage.value().getCustomer().getUsersInfo())
                            .stream()
                            .filter(Objects::nonNull)
                            .forEach(multiEmitter::emit);
                    multiEmitter.complete();
                })
                .onItem()
                .transformToUniAndMerge(party -> processParties(tokenId, (Party) party, userId))
                .toUni();
    }

    private Uni<Void> processParties(String tokenId, Party party, String userId) {
        /*
         * return type of RestA is Party
         */
        return callRestA(tokenId,party)
                .chain(res -> Uni.createFrom()
                        .item(this.updateAccounts(tokenId, party, party.getAccounts()))
                        .onFailure()
                        .invoke(failure -> LOGGER.error("Error while update accounts for userId - {} with stacktrace - {}", userId, failure.getMessage()))
                )
                .chain(partyRes -> {
                    List<Account> accounts = Optional.of(party)
                            .map(Party::getAccounts)
                            .orElse(List.of(new Account()));

                    return Multi.createFrom().emitter(accountsEmitter -> {
                        accounts.stream().filter(Objects::nonNull).forEach(accountsEmitter::emit);
                        accountsEmitter.complete();
                    })
                    .onItem()
                    .transformToUniAndMerge(account -> processRestBAndRestC(tokenId, userId,(Account) account))
                    .toUni()
                    .invoke(() -> LOGGER.info("processing complete for: {}", tokenId));
                });
    }

    private Uni<Void> processRestBAndRestC(String tokenId, String userId, Account account) {
        Uni<Boolean> processRestB = callRestB(tokenId, account, userId)
                                        .chain(res -> Uni.createFrom().item(this.updateRestB(tokenId,account.getFAId(),res)))

        Uni<Boolean> processRestC = callRestC(tokenId, account.getFAId(), account.getBId())
                                        .chain(res -> Uni.createFrom().item(this.updateRestC(tokenId, account.getFAId(), res, Constant.TopicLevel)))

        return Uni.join().all(callRestB, callRestC)
                    .andCollectFailures()
                    .invoke(() -> LOGGER.info(COMPLETED_PROCESSING_FOR_FA_INFO, account.getFAId()))
                    .replaceWithVoid();

    }

Both callRestB and callRestC return Uni<List> as its Mutiny and Quarkus based.

Now as I've explained, the issue is while using transformToUniAndMerge in the processParties method. What happens is sometimes even though the REST clients return the response it doesn't get mapped by the updateRestB and updateRestC methods. I don't care about the order my main purpose is that both these REST clients are called in parallel and asynchronously execute updateRestB and updateRestC so that the necessary information is mapped.

Using transformToUniAndConcatenate actually fixes my problem but the drawback is the total time increases as well which is a massive deal breaker.

What I want to achieve is as the items come from the MultiEmitter it async calls the REST clients and transforms them by calling the updateRestB and updateRestC. I don't care about the order, all I care about is at the end the information is mapped and I can't figure out a way to consistently map it using transformToUniAndMerge

[jponge]

I'm afraid it doesn't help much, but perhaps I wasn't clear.

What would be useful is a JUnit test, or a simple main method using only Mutiny APIs (and no Quarkus or Vert.x) that reproduces the problem.