quarkus: Quarkus Kafka Kotlin Coroutines does not work

Describe the bug

Hello to all. When trying to use Kotlin suspending functions together with Quarkus Kafka I am getting the following exception:

SRMSG00201: Error caught while processing a message: java.lang.IllegalStateException: No Vertx context found

I have simply defined the following code:

@Incoming("order-in")
    suspend fun consumeOrderEvents(orderRecord: Record<String, Order>) {

After analysing the issue I found that the code is not being called on a Vertx Thread as the exception is thrown from the following snippet:

val dispatcher: CoroutineDispatcher = Vertx.currentContext()?.let(::VertxDispatcher)
                ?: throw IllegalStateException("No Vertx context found")

in the AbstractSubscribingCoroutineInvoker class .

Is there anything I have to watch out for when configuring Quarkus Kafka and Kotlin ?

About this issue

Original URL
State: closed
Created 3 years ago
Reactions: 2
Comments: 20 (10 by maintainers)

Most upvoted comments

As far as I’m aware, this appears to be fixed and the issue can be closed. I didn’t open this issue, though.

lspahija on Mar 25, 2022

Looks good now with 2.8.0.CR1. Thank you and looking forward to the final release!

lspahija on Mar 25, 2022

Yes, Quarkus 2.8 depends on reactive messaging 3.15.0 which runs message handling methods on Vert.x context.

ozangunalp on Mar 25, 2022

We are now enforcing it in reactive messaging.

@ozangunalp was that merged in 2.8?

cescoffier on Mar 25, 2022

@cescoffier An identical exception is raised when using RabbitMQ connector.

This method:

@Incoming("my-channel")
suspend fun receiveUpdates(message: Message<String>) {
  // handle message
}

Produces this stacktrace:

java.lang.IllegalStateException: No Vertx context found at io.quarkus.smallrye.reactivemessaging.runtime.kotlin.AbstractSubscribingCoroutineInvoker.invoke(AbstractSubscribingCoroutineInvoker.kt:16) at io.quarkus.smallrye.reactivemessaging.runtime.kotlin.AbstractSubscribingCoroutineInvoker.invoke(AbstractSubscribingCoroutineInvoker.kt:11) at io.smallrye.reactive.messaging.AbstractMediator.invoke(AbstractMediator.java:88) at io.smallrye.reactive.messaging.SubscriberMediator.lambda$processMethodReturningACompletionStage$18(SubscriberMediator.java:216) at io.smallrye.context.impl.wrappers.SlowContextualSupplier.get(SlowContextualSupplier.java:21) at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage.subscribe(UniCreateFromCompletionStage.java:24) at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36) at io.smallrye.mutiny.operators.uni.UniOnItemOrFailureFlatMap.subscribe(UniOnItemOrFailureFlatMap.java:27) at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36) at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.request(UniToMultiPublisher.java:63) at io.smallrye.mutiny.operators.multi.MultiFlatMapOp$FlatMapInner.onSubscribe(MultiFlatMapOp.java:591) at io.smallrye.mutiny.converters.uni.UniToMultiPublisher.subscribe(UniToMultiPublisher.java:24) at io.smallrye.mutiny.groups.MultiCreate$1.subscribe(MultiCreate.java:156) at io.smallrye.mutiny.operators.multi.MultiFlatMapOp$FlatMapMainSubscriber.onItem(MultiFlatMapOp.java:191) at io.smallrye.mutiny.operators.multi.MultiMapOp$MapProcessor.onItem(MultiMapOp.java:50) at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61) at io.smallrye.mutiny.subscription.SafeSubscriber.onNext(SafeSubscriber.java:98) at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31) at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83) at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61) at io.smallrye.mutiny.operators.multi.MultiFlatMapOp$FlatMapMainSubscriber.tryEmit(MultiFlatMapOp.java:228) at io.smallrye.mutiny.operators.multi.MultiFlatMapOp$FlatMapInner.onItem(MultiFlatMapOp.java:597) at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61) at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.onItem(UniToMultiPublisher.java:82) at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:60) at io.smallrye.mutiny.operators.uni.builders.DefaultUniEmitter.complete(DefaultUniEmitter.java:36) at io.smallrye.mutiny.groups.UniOnNull.lambda$failWith$1(UniOnNull.java:43) at io.smallrye.context.impl.wrappers.SlowContextualBiConsumer.accept(SlowContextualBiConsumer.java:21) at io.smallrye.mutiny.groups.UniOnItem.lambda$transformToUni$4(UniOnItem.java:169) at io.smallrye.context.impl.wrappers.SlowContextualConsumer.accept(SlowContextualConsumer.java:21) at io.smallrye.mutiny.operators.uni.builders.UniCreateWithEmitter.subscribe(UniCreateWithEmitter.java:22) at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36) at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.performInnerSubscription(UniOnItemTransformToUni.java:81) at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:57) at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage$CompletionStageUniSubscription.forwardResult(UniCreateFromCompletionStage.java:63) at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) at io.quarkus.smallrye.reactivemessaging.runtime.devmode.DevModeSupportConnectorFactoryInterceptor.lambda$intercept$0(DevModeSupportConnectorFactoryInterceptor.java:42) at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) at io.quarkus.smallrye.reactivemessaging.runtime.devmode.ReactiveMessagingHotReplacementSetup$OnMessage$1.run(ReactiveMessagingHotReplacementSetup.java:44) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) at java.base/java.lang.Thread.run(Thread.java:833)

kdubb on Nov 15, 2021