@RunOnVertxContext and kotlin coroutines - awaitSuspending results in Thread blocked

[pkozakCC]

Hello.
I'm trying to write a test using @RunOnVertxContext, that executes suspending functions (R2DBC, Kafka etc).
Unfortunately, whenever Mutiny's awaitSuspending is executed, the vert.x thread is being blocked, resulting in a series of io.vertx.core.VertxException: Thread blocked exceptions being thrown (detailed stack trace below).
Below you can find a sample test that reproduces the error.
Interestingly, If the same code containing awaitSuspending is executed via HTTP request and triggered via RestAssured, everything works fine.

Test to reproduce:

    @Test
    @RunOnVertxContext
    fun suspendingTest() {
        runBlocking {
            // given
            val randomAlphabetic = randomAlphabetic(7)

            // when
            val result = Uni.createFrom()
                .emitter<String> { randomAlphabetic }
                .awaitSuspending()

            // then
            result shouldBeEqualTo randomAlphabetic
        }
    }

As junit tests cannot be suspending, I had to use runBlocking coroutine builder inside the test.
I suspect that this may be related to the issue, as the coroutines are not bound to the Vert.x instance managed by quarkus, however doing runBlocking(Vertx.currentContext().dispatcher()) won't solve it (dispatcher() from io.vertx:vertx-lang-kotlin-coroutines)

Stacktrace:

10:29:34.786 WARN   traceId=[] spanId=[] thread=[vertx-blocked-thread-checker] BlockedThreadChecker: Thread Thread[vert.x-eventloop-thread-0,5,main] has been blocked for 20437 ms, time limit is 2000 ms
io.vertx.core.VertxException: Thread blocked
	at java.base/jdk.internal.misc.Unsafe.park(Native Method)
	at java.base/java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:269)
	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:98)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:69)
	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:48)
	at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
	at app.channels.backend.adapters.user.UserFacadeIntTest.suspendingTest(UserFacadeIntTest.kt:40)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at io.quarkus.test.vertx.RunOnVertxContextTestMethodInvoker$RunTestMethodOnVertxEventLoopContextHandler.doRun(RunOnVertxContextTestMethodInvoker.java:189)
	at io.quarkus.test.vertx.RunOnVertxContextTestMethodInvoker$RunTestMethodOnVertxEventLoopContextHandler.handle(RunOnVertxContextTestMethodInvoker.java:178)
	at io.quarkus.test.vertx.RunOnVertxContextTestMethodInvoker$RunTestMethodOnVertxEventLoopContextHandler.handle(RunOnVertxContextTestMethodInvoker.java:149)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:270)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:252)
	at io.vertx.core.impl.ContextInternal.lambda$runOnContext$0(ContextInternal.java:50)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:994)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:1583)
10:29:36.791 WARN   traceId=[] spanId=[] thread=[vertx-blocked-thread-checker] BlockedThreadChecker: Thread Thread[vert.x-eventloop-thread-0,5,main] has been blocked for 22442 ms, time limit is 2000 ms
io.vertx.core.VertxException: Thread blocked
	at java.base/jdk.internal.misc.Unsafe.park(Native Method)
	at java.base/java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:269)
	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:98)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:69)
	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:48)
	at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
	at app.channels.backend.adapters.user.UserFacadeIntTest.suspendingTest(UserFacadeIntTest.kt:40)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at io.quarkus.test.vertx.RunOnVertxContextTestMethodInvoker$RunTestMethodOnVertxEventLoopContextHandler.doRun(RunOnVertxContextTestMethodInvoker.java:189)
	at io.quarkus.test.vertx.RunOnVertxContextTestMethodInvoker$RunTestMethodOnVertxEventLoopContextHandler.handle(RunOnVertxContextTestMethodInvoker.java:178)
	at io.quarkus.test.vertx.RunOnVertxContextTestMethodInvoker$RunTestMethodOnVertxEventLoopContextHandler.handle(RunOnVertxContextTestMethodInvoker.java:149)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:270)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:252)
	at io.vertx.core.impl.ContextInternal.lambda$runOnContext$0(ContextInternal.java:50)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:994)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:1583)

[mschorsch]

In your test, emitter.complete(...) is never called so emitter<String> blocks forever. If you change the test as follows it works:

    @Test
    @RunOnVertxContext
    fun suspendingTest() {
        runBlocking {
            // given
            val randomAlphabetic = randomAlphabetic(7)

            // when
            val result = Uni.createFrom()
                .emitter<String> { emitter ->
                    emitter.complete(randomAlphabetic)
                }.awaitSuspending()

            // then
            result shouldBeEqualTo randomAlphabetic
        }
    }

But I suppose that doesn't help. How about this:

    @Test
    fun suspendingTest2() {
        runSuspendingTest {
            // given
            val randomAlphabetic = randomAlphabetic(7)

            // when
            val result = Uni.createFrom()
                .emitter<String> { emitter ->
                    Thread.sleep(5000L)
                    emitter.complete(randomAlphabetic)
                }
                .awaitSuspending()

            // then
            result shouldBeEqualTo randomAlphabetic
        }
    }

    private fun runSuspendingTest(block: suspend () -> Unit) {
        VertxContextSupport.executeBlocking {
            runBlocking { block() }
        }.await().indefinitely()
    }

[pkozakCC]

Thank you for your response.
You are right, my reproduction test was faulty, however, I managed to fix my initial issue by upgrading quarkus from 3.15 LTS to 3.20 LTS, removing @RunOnVertxContext (as it was no longer available) and replacing it with VertxContextSupport.executeBlocking as you recommended.
Thank you.

[mschorsch]

@pkozakCC A small correction, @RunOnVertxContext is still available via the extension quarkus-test-vertx.