[Vertx] How to use a dedicated worker pool

[b-thiswatch]

Hi,

can anyone jump in and help me understand how to use a dedicated vertx pool, when consuming events? I have tried so many variations, but I simply don't get how it's supposed to be done.

I have a process that creates several thumbnails for images, stores them and creates a database entry.
Below are two ways that somehow looked promising, but they only utilise 1 thread of the thumbs worker.

@ApplicationScoped
class EventHandler(
    private val ensureThumbnails: EnsureThumbnails,
    vertx: Vertx,
) : Loggable {
    private val thumbsPool: WorkerExecutor =
        vertx.createSharedWorkerExecutor("thumbs", 4, 5, TimeUnit.MINUTES)

    @ConsumeEvent(FileEvents.ENSURE_THUMBS)
    fun process(cmd: EnsureThumbsCommand) {
        thumbsPool
            .executeBlockingAndForget {
                val ctx = Arc.container().requestContext()
                ctx.activate()
                ensureThumbnails.ensureThumbnails(cmd.fileReferenceId, cmd.wantedSizes)
                ctx.terminate()
            }
    }
}

Here I did not use @Blocking on the consume event, but then I had to activate the context myself.

@ApplicationScoped
class EventHandler(
    private val ensureThumbnails: EnsureThumbnails,
    vertx: Vertx,
) : Loggable {
    private val thumbsPool: WorkerExecutor =
        vertx.createSharedWorkerExecutor("thumbs", 4, 5, TimeUnit.MINUTES)

    @ConsumeEvent(FileEvents.ENSURE_THUMBS)
    @Blocking
    fun process(cmd: EnsureThumbsCommand) {
        thumbsPool
            .executeBlocking { 
                ensureThumbnails.ensureThumbnails(cmd.fileReferenceId, cmd.wantedSizes)
            }.await().indefinitely()
    }
}

This one looks a bit better for my taste, but also only utilises one thread of the thumbs pool.

Any help/advise is appreciated

Thanks

edit: Would something like this be a welcomed feature?

@ApplicationScoped
class EventHandler(
    private val ensureThumbnails: EnsureThumbnails,
    vertx: Vertx,
) : Loggable {
    private val thumbsPool: WorkerExecutor =
        vertx.createSharedWorkerExecutor("thumbs", 4, 5, TimeUnit.MINUTES)

    @ConsumeEvent(FileEvents.ENSURE_THUMBS)
    @WorkExecutor(name = "thumbs") // <-- you get the idea...
    @Blocking
    fun process(cmd: EnsureThumbsCommand) = ensureThumbnails.ensureThumbnails(cmd.fileReferenceId, cmd.wantedSizes)

}

[quarkus-bot[bot]]

/cc @geoand (kotlin)

[geoand]

cc @cescoffier

[mschorsch]

You could try using a ManagedExecutor (https://quarkus.io/guides/context-propagation#usage-example-for-completionstage) instead of the Vert.x worker pool. The ManagedExecutor propagates the context.

[cescoffier]

The simplest with Java 21 is to use @RunOnVirtualThread.
If you want to use your own executor, the simplest is to dispatch the work to that worker thread. If there is a "result" to produce, do something like:

@ConsumeEvent("...")
public Uni<String> process(String input) {
    return Uni.createFrom().item(input)
         .emitOn(myExecutor)
         .map(i -> doTheWork(i)); // Will be called on a thread of the executor        
}

[b-thiswatch]

@cescoffier thanks! Will check out the emitOn
I have tried @RunOnVirtualThread but end up with only Unable to acquire JDBC Connection errors.

[cescoffier]

ah, that's interesting. How much concurrency do you have?

[b-thiswatch]

I'm pushing ~450k events and the system is running on 10 cores. But the Unable to acquire JDBC Connection happens on the first execution already.

Or did I misunderstand?

[b-thiswatch]

Sorry, no, that was wrong. It does not fail immediately, but rather really quickly. I will try to capture

[b-thiswatch]

Maybe, I should rephrase the question. What is the easiest way to add and use a second worker pool (with a different number of workers and different timeouts) to a Quarkus App?

[cescoffier]

You can just define your own scheduler, but be aware that context propagation may not be supported.