Quarkus Reactive flow execution while calling External Service

[nitinty]

Hi Team,

I'm trying to understand reactive flow execution in Quarkus and have a few doubts. I believe this is the best platform to ask.

I have an application where we call an external service and use its response within the calling service. As I understand, in reactive programming, code executes on an event loop thread by default. However, we can't block the event loop thread for long-running operations since the number of event loop threads is limited. Such processing should instead be offloaded to a worker thread.

I created the following method to manually run the external service call on a worker thread. I just wanted to confirm: Is this the correct way to handle such operations, or does Quarkus provide any mechanism to automatically detect blocking operations and switch to a worker thread?

Please assist here, Thanks in advance

package org.acme.demo.service;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.core.Response;
import lombok.extern.slf4j.Slf4j;
import org.acme.demo.entity.EmployeeEntity;
import org.acme.demo.model.Employee;
import org.acme.demo.out.IDepartmentController;
import org.acme.demo.repository.EmployeeRepositoryImpl;
import org.eclipse.microprofile.rest.client.inject.RestClient;


@ApplicationScoped
@Slf4j
public class EmployeeService {

    @Inject
    EmployeeRepositoryImpl employeeRepository;

    @Inject
    @RestClient
    IDepartmentController departmentController;


    public Uni<Response> createEmployee(Employee employee) {

        // Offload the external service call to a worker thread
        return Uni.createFrom()
                .deferred(() -> {
                    log.info("1 EmployeeService :: create with thread {}", Thread.currentThread().getName());
                    return departmentController.getDepartmentDetails(); // External Service call, This call taking 5 (as I added Thread.sleep() intentionally)
                }).runSubscriptionOn(Infrastructure.getDefaultWorkerPool())  // Offload to worker thread
                .onFailure().transform(throwable -> {
                    throwable.printStackTrace();
                    return new RuntimeException("Failed to fetch department details", throwable);
                }).chain(department -> {
                    // Use the result from the external service
                    log.info("2 EmployeeService :: create with thread {}", Thread.currentThread().getName());
                    EmployeeEntity employeeEntity = EmployeeEntity.builder()
                            .firstname(employee.getFirstname())
                            .lastname(employee.getLastname())
                            .password(employee.getPassword())
                            .departmentName(department.getName())
                            .build();

                    // Persist employee in the database
                    return employeeRepository.persist(employeeEntity);
                }).map(savedEmployee -> Response.ok(savedEmployee).build());
    }
}

---

I also found some informative blogs

https://quarkus.io/blog/io-thread-benchmark/#imperative-and-reactive-the-elevator-pitch
https://quarkus.io/blog/resteasy-reactive-faq/
https://quarkus.io/blog/resteasy-reactive-smart-dispatch/

[nitinty]

@cescoffier @geoand @sberyozkin
please assist here

[nitinty]

DB interaction are made properly async

I'm using ReactivePanacheMongoRepository in my repository layer. Isn't this the correct approach ?

package org.acme.demo.repository;

import io.quarkus.mongodb.panache.reactive.ReactivePanacheMongoRepository;
import io.smallrye.mutiny.Uni;
import org.acme.demo.entity.EmployeeEntity;

import java.util.List;

public interface EmployeeRepository extends ReactivePanacheMongoRepository<EmployeeEntity> {

    Uni<EmployeeEntity> persistEmployee(EmployeeEntity employeeEntity);
}
`

```java
package org.acme.demo.repository;

import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import org.acme.demo.entity.EmployeeEntity;

import java.util.List;

@ApplicationScoped
public class EmployeeRepositoryImpl implements EmployeeRepository {
    
    public Uni<EmployeeEntity> persistEmployee(EmployeeEntity employeeEntity) {
        return persist(employeeEntity);
    }


}

[geoand]

I'm using ReactivePanacheMongoRepository in my repository layer. Isn't this the correct approach ?

It is!

[nitinty]

Can you please have a look on above mentioned point as well ?

#43951 (reply in thread)

[nitinty]

I'm using ReactivePanacheMongoRepository in my repository layer. Isn't this the correct approach ?

It is!

One more thing: when I put a debugger in the MongoDB library code and print Thread.currentThread().getName(), it always print the event loop thread (that means mongo code is running in event loop thread). I'm a bit confused, Does it ever switch to a worker thread for database processing when using MongoDB with Panache, or is that something the developer needs to handle manually ?

[geoand]

Is this the correct approach, or is there a mechanism in Quarkus that can automatically detect and offload blocking operations to a worker thread?

There is not. If an operation is blocking, you need to make sure you are not running it an event loop thread - that will be catastrophic for performance under even moderate load