Uni runSubscription on and Timeouts

[krezchikov42]

Hello there! I am a little confused on how to create rest level timeouts when using quarkus-resteasy-reactive

This is a an example of a java class that I have. Instead of thread.sleep I have a blocking database call. Since it is blocking according to docs we use runSubscriptionOn. However, I've noticed through testing that this does not actually make it so the whole response will timeout after 150 ms. If you send multiple requests to this endpoint, they will all pass. This is because the the .ifNoItem().after(Duration.ofMillis(150)) starts it's timer from when it is subscribed to on the singleThreadPool. So despite being in the queue for the singleThreadPool for longer than 150ms the requests won't time out. However, I want it so that if any response takes longer than 150 ms then it should timeout. So if you sent 4 requests at the same time to this endpoint, the first one would make it, and the next 3 would time out. I want to make it so that if the request has been waiting in the executor queue for longer than 150ms then the response will timeout and it will not make a request to the DB.
Notes, I have to use my own executor to limit the number of concurrent queries being made to the db.

I think I have a slight misunderstanding of how to use quarkus-rest and how to use it with blocking database calls. Should I not be returning a Uni from my resource?

I'm using quarkus 3.8.4 and quarkus-resteasy-reactive

 @GET
    @Path("/testNewTimeout")
    @Produces(MediaType.APPLICATION_JSON)
    public Uni<Response> testNewTimeout(@QueryParam("number") @DefaultValue("1") int number) {
        return createUniNewTimeout(number);
    }

    private Uni<Response> createUni(int number) {
        long subscriptionTime = System.currentTimeMillis();
        return Uni.createFrom().item(() -> {
            long executionStartTime = System.currentTimeMillis();
            long queueTime = executionStartTime - subscriptionTime;
            System.out.println("Uni " + number + " - Queue time: " + queueTime + "ms, Starting execution");
            try {
                Thread.sleep(100);  //substitution for blocking database call
                System.out.println("Uni " + number + " - Completed execution");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return Response.status(Response.Status.ACCEPTED).entity("Accepted").build();
        })
            .runSubscriptionOn(singleThreadPool)
            .ifNoItem()
            .after(Duration.ofMillis(150))
            .fail()
            .onFailure()
            .invoke(throwable -> {
                long failureTime = System.currentTimeMillis();
                long totalTime = failureTime - subscriptionTime;
                System.out.println("Uni " + number + " - FAILED after " + totalTime + "ms total: " + throwable.getClass().getSimpleName());
            })
            .onFailure()
            .recoverWithItem(throwable -> Response.status(Response.Status.REQUEST_TIMEOUT)
                .entity("Request timed out")
                .build()
            );

    }
    ```

[cescoffier]

Why do you use reactive if you have a blocking database?

Also, the ifNoItem starts when the subscription is received, which, in your case, is delayed by the scheduler queue and item emission. So, the code behaves correctly.

To get it working, you should not block item emission (which runs on the subscription thread in your case). Using an emitter should fix it (it will require a bit of work, since in the emitter callback, you need to start your thread (instead of runOnSubscription) and emit the item when ready).

[krezchikov42]

So basically if we are not going to go the reactive route, we should stay away from Unis correct?

[jponge]

Uni is useful if you are doing async I/O.

You can always use a Uni from a blocking context (see uni.await().(...)) but it's not the primary use-case.

[cescoffier]

As @jponge said: if you use non-blocking I/O (reactive database driver, non blocking rest/http client...), Unis make sense, but here you are using a blocking database driver.

The question is: how much concurrency do you want to achieve?

[krezchikov42]

I mean, isn't submitting a task to an executor to make a database call make it async? Just like running a blocking database call with runSubscriptionOn and returning a Uni in the response will make it async? I mean thats what this documentation says

Uni<String> uni = Uni.createFrom()
        .item(this::invokeRemoteServiceUsingBlockingIO)
        .runSubscriptionOn(Infrastructure.getDefaultWorkerPool());

Is the code below async? Yes because we use an emitter and pass back a value to the emitter. Would you recommend this sort of approach if we want to hit database that have blocking calls and also institute serverside timeouts if operations run too long?

 private Uni<Void> createUni(int number) {
        return Uni.createFrom().emitter(emitter -> {
            long subscriptionTime = System.currentTimeMillis();
            var future = singleThreadPool.submit(() -> {
                long executionStartTime = System.currentTimeMillis();
                long queueTime = executionStartTime - subscriptionTime;
                System.out.println("Uni " + number + " - Queue time: " + queueTime + "ms, Starting execution");
                try {
                    Thread.sleep(100);
                    System.out.println("Uni " + number + " - Completed execution");
                    emitter.complete(null);
                } catch (InterruptedException e) {
                    System.out.println("Uni " + number + " - Interrupted!");
                    emitter.fail(e);
                }
            });

            // Cancel the task if the Uni is cancelled
            emitter.onTermination(() -> {
                System.out.println("Uni " + number + " - Termination triggered, cancelling future");
                future.cancel(true);
            });
        })
            .ifNoItem()
            .after(Duration.ofMillis(150))
            .fail()
            .onFailure()
            .invoke(throwable -> {
                System.out.println("Uni " + number + " - FAILED: " + throwable.getClass().getSimpleName());
            })
            .replaceWithVoid();
    }

[jponge]

The goal of using async I/O is to multiplex concurrent request processing to a single (event-loop) thread, so you can handle more concurrent connections with fewer resources.

If you just offload blocking tasks to an executor of worker threads you still have a hard limit to your concurrency level, and that limit is the executor pool size, which effectively can't be better than using worker threads from the start.