Quarkus + Mutiny Kotlin - io.vertx.core.VertxException: Thread blocked

[pwlan]

Context

Hi,
I am devleoping a simple Reactive REST + Mongo service that uses quarkus, kotlin coroutines and mutiny-kotlin.

Description

When converting a Multi to a koltin Flow within the rest endpoint, i get an io.vertx.core.VertxException: Thread blocked exception.

Example code:

    @GET
    suspend fun getData(): Data?  {
        val collection: ReactiveMongoCollection<Data> = getCollection()
        return collection.find().asFlow().firstOrNull()
    }

Calling this rest endpoint yields the following exception and is then seemingly blocked indefinitely.

11:30:02 WARN  [io.ve.co.im.BlockedThreadChecker] (vertx-blocked-thread-checker) [] Thread Thread[vert.x-eventloop-thread-28,5,main] has been blocked for 3530 ms, time limit is 2000 ms: io.vertx.core.VertxException: Thread blocked
	at [email protected]/jdk.internal.misc.Unsafe.park(Native Method)
	at [email protected]/java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:252)
	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:88)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:59)
	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
	at io.smallrye.mutiny.coroutines.MultiKt$asFlow$2$subscription$1.accept(Multi.kt:32)
	at io.smallrye.mutiny.subscription.Subscribers$CallbackBasedSubscriber.onItem(Subscribers.java:67)
	at io.smallrye.mutiny.operators.multi.MultiEmitOnOp$MultiEmitOnProcessor.run(MultiEmitOnOp.java:199)
	at io.quarkus.mongodb.impl.Wrappers.lambda$toMulti$2(Wrappers.java:30)
	at io.quarkus.mongodb.impl.Wrappers$$Lambda$1618/0x00000008015ec740.handle(Unknown Source)
	at io.vertx.core.impl.AbstractContext.dispatch(AbstractContext.java:100)
	at io.vertx.core.impl.AbstractContext.dispatch(AbstractContext.java:63)
	at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:38)
	at io.vertx.core.impl.EventLoopContext$$Lambda$1378/0x000000080145c940.run(Unknown Source)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at [email protected]/java.lang.Thread.run(Thread.java:831)

I assume it may be the call to runBlocking in the MultiKt$asFlow function.

Thanks for any help on the matter.

Additional details

java11
mutiny-kotlin:1.1.2 and 1.0.0
quarkus:2.2.3.Final
kotlin:1.5.21

[jponge]

/cc @heubeck

[heubeck]

Thank you @pwlan for reporting this issue, hope to find some time for looking into it.
Could you maybe in the meanwhile try to create a reproducer in form of test methods?

There are general issues when having embedded coroutine scopes on the same dispatcher: https://youtrack.jetbrains.com/issue/KT-44245 (just collecting some random thoughts).
There was no Coroutine support in ResteasyReactive at the time when mutiny-kotlin was started, so maybe there are interferences with the dispatchers .

Short explaination about runBlocking in Multi.asFlow (if helpful for others thinking about this issue): The Flow.emit should be synchronous (blocking) executed on that Thread, that executes Subscriber.onItem to make consumers of the Flow behave the same, as if directly used within onItem - hope not to have a fallacy in that idea...

Edit: Having said this, of course a consumer of the Flow´ can block the onItemsubmission - like anyonItemimplementation could do (but for sure must never block _forever_ as in the issue description). But: Could it make sense to decouple theMultiemission and theFlowconsumption using a bufferedFlow`? WDYT?

Maybe @evanchooly as an expert in Coroutine stuff can quickly verify, if there's a conceptual mistake in https://github.com/smallrye/smallrye-mutiny/blob/main/kotlin/src/main/kotlin/io/smallrye/mutiny/coroutines/Multi.kt#L25 ? 🙏

[heubeck]

@pwlan, I've found a bug, that could cause your finding.
Could you maybe do me a favor, and test with a (local) build of the head branch of #734 ?

[pwlan]

Wow, that was quick. Sure, i can give it a test later today.

[pwlan]

Ok, i gave the fix a test. But it does not seem to fix this issue.

What did I do:

1. downloaded the master branch containig the fix.
2. built and installed it to the my local maven repo.
3. created a simple junit test (see attachment mutiny-kotlin-test.zip)
4. ran test.

Attachment
mutiny-kotlin-test.zip

I hope this helps.

PS: I can mitigate the issue by simply changing the dispatcher.
For example,

withContext(Dispatchers.Unconfined) {
   myMulti.asFlow().first()
}

[heubeck]

Hey @pwlan ,
fix will be contained in the next release.
Thank you for reporting, and happy use of the mutiny-kotlin module.

[pwlan]

Sounds great. Thanks for all the effort and good work!!