Blocking transaction inside websocket

[salemsd]

Hello,

I’m using Quarkus WebSockets Next extension and I’m running into issues with transactional database operations.

• I'm getting a BlockingOperationNotAllowedException as soon as I use any method annotated with @transactional. I found a similar post here that suggested to add quarkus.websocket.dispatch-to-worker=true to application.properties to move WebSocket handling to worker threads, but I think this doesn't apply to websocket next

I have tried using .runSubscriptionOn(Infrastructure.getDefaultWorkerPool()), but it doesn't change anything.

Thank you in advance for your help.

[mkouba]

Hello @salemsd, could you share your code, or at least a relevant code snippet? BlockingOperationNotAllowedException is thrown when you execute a blocking operation on IO thread (event loop). There are some well-defined rules to determine the execution model for a WS Next endpoint callback method.

There is also io.quarkus.vertx.VertxContextSupport#executeBlocking(Callable<T>) (in Quarkus 3.16+) that can be used to execute some blocking code as part of a reactive pipeline.

[salemsd]

Hello,

This is one of the methods in my web socket that uses a transactional operation:

  private Multi<String> streamAndRetry(Multi<String> response, StringBuilder fullMessageBuilder) {
    return Multi.createFrom().emitter(emitter -> {
      response.onItem().invoke(fullMessageBuilder::append)
              .onCompletion().call(Unchecked.supplier(() -> tryCompile(fullMessageBuilder)))
              .onFailure().invoke(failure -> emitter.emit("fail"))
              .onFailure().retry().atMost(3)
              .onFailure().recoverWithMulti(() -> {
                var failureMessage = "There was an error while compiling the code. Please try again.";
                return saveMessageReactive("assistant", failureMessage)
                        .onItem().transformToMulti(ignored -> Multi.createFrom().item(failureMessage));
              })
              .subscribe().with(emitter::emit, emitter::fail);
    });
  }

  private Uni<Void> saveMessageReactive(String emitter, String messageContent) {
    return chatService.saveMessage(activeConversation, emitter, messageContent)
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
  }

  @Transactional
  public Uni<Void> saveMessage(Conversation conversation, String emitter, String messageContent) {
    System.out.println("Saving message");
    return Uni.createFrom().voidItem()
            .onItem().invoke(() -> {
              var chatMessage = Chat.create(conversation, emitter, messageContent, LocalDateTime.now())
              System.out.println("Saving message: " + chatMessage);
              entityManager.persist(chatMessage);
              conversation.setLastModified(chatMessage.getTimestamp());
              entityManager.merge(conversation);
              System.out.println("Message saved");
            });
  }

saveMessage is one of the methods that trigger the exception.

I'm using Hibernate ORM and Jakarta Persistence, with sqlite jdbc driver. I unfortunately cannot use Quarkus 3.16 as I'm required to use an older version (3.14.2) in the context of my project.

[mkouba]

Well, I don't think it's a good idea to mix the reactive pipelines (Mutiny) together with JTA (@Transactional), which is inherently a blocking technology. It will be inefficient at the very least.

You can annotate the WS endpoint callback method with @io.smallrye.common.annotation.Blocking in order to instruct Quarkus to execute the method on a worker thread. Alternatively, just copy the logic from VertxContextSupport#executeBlocking(Callable<T>) and wrap the invocation of saveMessage(). Finally, you can also consider using Hibernate Reactive API .

It seems that Quarkus REST has a specific rule for @Transactional: "If a method or class is annotated with jakarta.transaction.Transactional then it will also be treated as a blocking method. This is because JTA is a blocking technology, and is generally used with other blocking technology such as Hibernate and JDBC." Maybe we should do the same in WS Next? WDYT @cescoffier

[cescoffier]

I agree we should do the same in WS-Next and in Quarkus Langchain4J where I ... forgot this case.

[mkouba]

I feel like it's a little uneducative but still: #44495 ;-)

[salemsd]

Thank you. I managed to manually run the operations through an executor service and it worked. Might not be the best solution but I was running out of time. Using the @Blocking annotation was not working and switching to Hibernate Reactive wasn't an option in the context of my school project. I marked the discussion as answered.

[mkouba]

Using the @Blocking annotation was not working

Could you share some details about what exactly was not working? Thanks.