BlockingOperationNotAllowedException - You have attempted to perform a blocking operation on a IO thread

Question

I'm making a GRPC service in my quarkus app and when I try to make a request on my method I get this error:

UNKNOWN: io.quarkus.runtime.BlockingOperationNotAllowedException - You have attempted to perform a blocking operation on a IO thread. This is not allowed, as blocking the IO thread will cause major performance issues with your application. If you want to perform blocking EntityManager operations make sure you are doing it from a worker thread.

So I read in this article that I should add the annotation @Blocking to use the worker thread.

After putting it, i got the same error.

There is my code:

GRPCService

    @Override
    @Blocking
    public Uni<MerchantGRPC> findById(idMerchantGRPC request) {

        // Find in database the merchant (getting error in this)
        MerchantDTO merchantDTO =     merchantService.findById(request.getId());

        return Uni.createFrom().item(fromMerchantDTOtoMerchantGRPC(merchantDTO));
    }

Edit

Service in my proto

    service Greeter {
      rpc SayHello (HelloRequest) returns (HelloReply) {}
      rpc findById (idMerchantGRPC) returns (MerchantGRPC) {}
    }

Dependencies

    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-arc</artifactId>
        </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-resteasy-reactive</artifactId>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-junit5</artifactId>
          <scope>test</scope>
    </dependency>
    <dependency>
          <groupId>io.rest-assured</groupId>
          <artifactId>rest-assured</artifactId>
          <scope>test</scope>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-resteasy-reactive</artifactId>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-smallrye-openapi</artifactId>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-hibernate-orm-panache</artifactId>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-jdbc-mysql</artifactId>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-resteasy-reactive-jackson</artifactId>
    </dependency>
    <dependency>
          <groupId>org.mapstruct</groupId>
          <artifactId>mapstruct</artifactId>
          <version>1.5.2.Final</version>
    </dependency>
    <dependency>
          <groupId>org.mapstruct</groupId>
          <artifactId>mapstruct-processor</artifactId>
          <version>1.5.2.Final</version>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-hibernate-validator</artifactId>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-junit5-mockito</artifactId>
          <scope>test</scope>
    </dependency>
    <dependency>
          <groupId>io.rest-assured</groupId>
          <artifactId>rest-assured</artifactId>
          <scope>test</scope>
    </dependency>
    <dependency>
          <groupId>io.quarkus</groupId>
          <artifactId>quarkus-grpc</artifactId>
    </dependency>

Do you have any idea that could help ?

Answer 1

To use Uni<?> you need to emmit to a executor, to run in a work-thread and not in a main event loop thread, see a sample below.

product.proto:

service ProductService {
  rpc FindById(FindRequest) returns (ProductDTO);
  rpc CreateProduct(CreateRequest) returns (ProductDTO);
}

message FindRequest {
  string id = 1;
}

message CreateRequest {
  string name = 1;
}

message ProductDTO {
  string id = 1;
  string name = 2;
}

ProductRepository.java

@ApplicationScoped
public class ProductRepository implements PanacheRepositoryBase<ProductEntity, Long> {
}

MyProductService.java

@ApplicationScoped
public class MyProductService {

    @Inject
    ProductRepository repository;

    @Transactional
    public ProductEntity create(String name) {
        Log.infov("into create");
        var pe = new ProductEntity();
        pe.setName(name);
        repository.persist(pe);
        return pe;
    }

    public ProductEntity findById(Long id) {
        return repository.findById(id);
    }

}

GRPCProductService.java

@GrpcService
public class GRPCProductService implements ProductService {



    @Inject
    MyProductService service;

    private ExecutorService executor = Executors.newFixedThreadPool(4);

    @Override
    public Uni<Product.ProductDTO> findById(Product.FindRequest request) {
        return Uni.createFrom()
                .item(request)
                .emitOn(executor) //   <-- HERE
                .map((r) -> {
                    var entity = service.findById(Long.parseLong(r.getId()));
                    return Product.ProductDTO.newBuilder()
                            .setId(entity.getId().toString())
                            .setName(entity.getName())
                            .build();
                });
    }

    @Override
    public Uni<Product.ProductDTO> createProduct(Product.CreateRequest request) {
        Log.infov("into createProduct");
        return Uni.createFrom()
                .item(request)
                .emitOn(executor) //   <-- HERE
                .map(r -> {
                    Log.infov("into createProduct map");
                    var entity = service.create(r.getName());
                    return Product.ProductDTO.newBuilder()
                            .setId(entity.getId().toString())
                            .setName(entity.getName())
                            .build();
                });
    }
}

log output from this sample:

2023-10-18 13:16:33,082 INFO  [dem.pro.GRPCProductService] (vert.x-eventloop-thread-0) into createProduct
2023-10-18 13:16:33,084 INFO  [dem.pro.GRPCProductService] (pool-7-thread-1) into createProduct map
2023-10-18 13:16:33,099 INFO  [dem.pro.MyProductService] (pool-7-thread-1) into create

Answer 2

So I decided to use the default gRPC API instead of the Mutiny API because @Blocked don't seems to change the thread to worker pool for my method (Blocked usage). After some changings I got the same error but it works when i used @Blocked on the method (this time the annotation seems to do his job).

Answer 3

There is no goot reason do use a Uni in this case. It would be far easier to just do:

@Override
public MerchantGRPC findById(idMerchantGRPC request) {

    // Find in database the merchant (getting error in this)
    MerchantDTO merchantDTO = merchantService.findById(request.getId());

    return fromMerchantDTOtoMerchantGRPC(merchantDTO);
}