Quarkus reactive panache how to use @ReactiveTransactional annotation in non-REST endpoint

Question

I have a service that regularly receives messages in a batch, and then for each of those messages perform a transaction.

Here is the sample code:

// scheduled to run every 5 second
private Uni<Void> receiveMessage() {
    Uni<List<Message>> uniListMessage = messageService.get();
    
    return uniListMessage
            .map(listMessage -> {
                final UniJoin.Builder<Void> uniBuilder = Uni.join().builder();
                listMessage.forEach(message -> {
                    Uni<Void> uniOperation = businessService.doOperation(message);
                    uniBuilder.add(uniOperation);
                });
                
                return uniBuilder.joinAll().andCollectFailures()
                        .replaceWithVoid();
            });
}

And in BusinessService:

@ReactiveTransactional
public Uni<Void> doOperation(Message message) {
    // do find query, insert, update, etc to multiple tables
    Uni<Fruit> uniFruit = fruitRepo.findByMessage(message);
    
    return uniFruit
            .map(fruit -> {
                // calculate things
                final UniJoin.Builder<Void> uniBuilder = Uni.join().builder();
                
                Uni<Apple> uniApple = appleRepo.doUpdate(fruit);
                Uni<Melon> uniMelon = melonRepo.doUpdate(fruit);

                return uniJoinBuilder.joinAll().andFailFast();
            });
}

appleRepo.doUpdate() and melonRepo.doUpdate does not have @ReactiveTransactional annotation, and only doing .persist() or .update() using PanacheRepositoryBase

I'm aware they recommend us to put @ReactiveTransactional at REST endpoint controllers, but I'm not using REST so I'm not sure how to do this correctly

I run some test, if I just commit 1 message everything works fine

However things starting to get weird if I have ~100 messages, for example I got:

NoStackTraceThrowable: Transaction already completed
java.lang.IllegalStateException: HR000061: Session is currently connecting to database
io.vertx.mysqlclient.MySQLBatchException: Error occurs during batch execution

and very rarely

index out of bound 1 out of 0

I previously using classic-orm, but got into trouble because the @Transactional cannot be used for function that returns Uni