relative how to trigger a background/side job without waiting it

[jie-huang]

quarkus reactive uses mutiny to handle task asynchronously.
But, a normal flow is always wait every jobs to finish, then returns the result.
Sometime, I just want to trigger a job and let it run in the background without waiting it to be done.
Any suggestion or example?

Uni<Integer> mainJob() {
  return Uni.createFrom().item(1);   // fake logic
}

Uni<Void> sideJob(int n) {
  logger.log("result = " + n);   // fake logic
}

@Path("test")
Uni<Integer> testExample() {
    return mainJob().onItem().call(n -> sideJob(n)); // question place, it still wait sideJob() to be done first.
}

There are two ideas from my understanding,

1. Using ManagedExecutor

@Inject ManagedExecutor executor;

@Path("test")
Uni<Integer> testExample() {
    return mainJob().onItem().invoke(n -> executor.runAsync(n -> sideJob(n)));
}

2. Using Uni.subscribe()

@Path("test")
Uni<Integer> testExample() {
    return mainJob().onItem().invoke(n -> sideJob(n).subscribe().with(
      result -> System.out.println("done"),
      failure -> System.out.println(failure)));
}

What is the suggested solution for this use case in Quarkus?
Or, with the subscribe() way, does the context work?
e.g. sideJob() may use injected objects, when mainJob result is returned to client, are those objects still available to sideJob() (it should still run after the response has been sent to client and may last several minutes)?

Thanks

[mkouba]

1. Using ManagedExecutor

This should work.

2. Using Uni.subscribe()

I'm not 100% sure but I don't think this would work because sideJob(n).subscribe() is invoked on the thread emitting the event and blocks until the sideJob(n) is completed.

You might need something like sideJob(n).runSubscriptionOn(executor).subscribe().

An alternative might be to inject the Vertx instance and use either runOnContext() or executeBlocking().

CC @cescoffier

e.g. sideJob() may use injected objects, when mainJob result is returned to client, are those objects still available to sideJob() (it should still run after the response has been sent to client and may last several minutes)?

It depends on the scope of the injected bean. For example, it's always safe to use @ApplicationScoped, @Singleton and @Dependent beans but @RequestScoped beans (1) must be propagated to the other thread and (2) are destroyed when the request is completed.

[jie-huang]

@mkouba
I tried your suggestion, if I do sideJob(n).runSubscriptionOn(executor).subscribe(), it reports error like

java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [488]: 'vert.x-eventloop-thread-4' current Thread [479]: 'executor-thread-0'

Without runSubscriptionOn(), it works.

So, quarkus requires session on the same thread. Is that right?

[mkouba]

@mkouba I tried your suggestion, if I do sideJob(n).runSubscriptionOn(executor).subscribe(), it reports error like

java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [488]: 'vert.x-eventloop-thread-4' current Thread [479]: 'executor-thread-0'

Without runSubscriptionOn(), it works.

So, quarkus requires session on the same thread. Is that right?

Not Quarkus itself but the Hibernate Reactive does. A reactive session may not be accessed concurrently.

[DevEnol]

Hi!

Try using something like this pattern:

@Path("test")
Uni<Integer> testExample() {
    sideJob(n).runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .subscribeAsCompletionStage()
            .whenComplete((i, e) -> {
                if(e != null) {
                    logger.error(e.getLocalizedMessage());
                }
            });

    return mainJob();
}

Uni<Void> sideJob(int n) {
    return Uni.createFrom().voidItem()
            .onItem().delayIt().by(Duration.ofSeconds(2))    // simulate delay for 2 second
            .map(v -> {
                logger.log("result " + n);
                return v;
            });
}

[jie-huang]

I got the same error of

java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [496]: 'vert.x-eventloop-thread-3' current Thread [488]: 'executor-thread-0'

In the sideJob, it uses reactive panache. @mkouba is right, it seems the hibernate must be run in the same thread.

And, the weird part now is, if I use

sideJob().subscribe().with(r -> logger.info("done"), failure -> logger.error("failed"));

It works well in unittest.
But it fails when running the app, the really job of sideJob() (reactive panache) just hangs.

[DevEnol]

@jie-huang I am using Reactive SQL Client with no issue.

[jie-huang]

@DevEnol
I created a demo project at https://github.com/jie-huang/quarkus-reactive-panache

Could you please take a look?
Thanks

[DevEnol]

I have already test it with Hibernate Reactive Panache extension, with the same error result:

Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [496]: 'vert.x-eventloop-thread-3' current Thread [488]: 'executor-thread-0'

I just want to inform you for using Reactive SQL Client extension if possible as an alternative.

[jie-huang]

@DevEnol
Thanks for the suggestion.