Virtual Threads when consuming Events

[zthreee]

Hey all,

I am just starting to learn Quarkus and trying to understand some of its functionalities. I am rewriting my Spring app, to see what Quarkus has to offer, but I stumbled on an issue, where I cannot find an answer for.

So I have a consumer and publisher (via EventBus - vertx) where i publish around 3-5k messages per minute.

I tried to write that in this way:

    @ConsumeEvent("kline.events")
    public void consume(Kline klineEvent) {
        Thread.startVirtualThread(() -> {
            log.debug("Received kline event: {}", klineEvent);
            try {
                var currentState = stateManager.getSymbolState(klineEvent.symbol());
                switch (currentState) {
                    case NOT_INITIALIZED -> handleNotInitialized(klineEvent);
                    case INITIALIZING -> handleInitializing(klineEvent);
                    case READY -> handleReady(klineEvent);
                }
            } catch (Exception e) {
                log.error("Error processing kline: {}", klineEvent, e);
            }
        });
    }

But somehow, this piece of code, after some time (like 4-5 minutes) just stops receiving the events - even though they are published, there is no error, no warning, just stops and thats it.

Tried to debug it and learn about Quarkus and VirtualThreads and instead creating the virtualThread like I did above, i just added RunOnVirutalThread ->

    @RunOnVirtualThread
    @ConsumeEvent("kline.events")
    public void consume(Kline klineEvent) {
        log.debug("Received kline event: {}", klineEvent);
        try {
            var currentState = stateManager.getSymbolState(klineEvent.symbol());
            switch (currentState) {
                case NOT_INITIALIZED -> handleNotInitialized(klineEvent);
                case INITIALIZING -> handleInitializing(klineEvent);
                case READY -> handleReady(klineEvent);
            }
        } catch (Exception e) {
            log.error("Error processing kline: {}", klineEvent, e);
        }
    }

And that works perfectly, at this moment I am keeping uptime of several hours and nothing is being missed.

Great that it works, but can anyone explain, why the first solution is wrong when it comes to quarkus and its programming? Why the consumer stopped receiving events?

[quarkus-bot[bot]]

/cc @cescoffier (virtual-threads), @ozangunalp (virtual-threads)

[cescoffier]

@RunOnVirtualThread uses a specific virtual thread scheduler (that you can inject and use directly), which does a bit of magic to propagate context (and also.... give names to virtual threads).

See:

• https://quarkus.io/guides/virtual-threads#use-the-duplicated-context-in-virtual-threads
• https://quarkus.io/guides/virtual-threads#virtual-thread-names
• https://quarkus.io/guides/virtual-threads#inject-the-virtual-thread-executor

[zthreee]

Hmm, but that does not answer my question, why the first option is randomly stopping working after few minutes - is there any specific reason that quarkus just somehow kills that?

[cescoffier]

We are a bit safer when we enqueue the task. But, it's hard to say. We recommend using our executor if you want to spawn virtual thread yourself.