Why do I need to provide a dummy @Outgoing channel to have a terminal @Incoming consumer?

[erplsf]

Greetings,

I'm getting started with smallrye-reactive-messaging and encountered one annoyance that I think should not be present.
I want to create a "terminal" consumer of Multi<T> - so I can batch-consume multiple payloads from my channel.
To achieve that I try to write the simplest code possible:

package playground.kafka

import jakarta.enterprise.context.ApplicationScoped

import java.time.Duration

import org.eclipse.microprofile.reactive.messaging.Incoming
import org.eclipse.microprofile.reactive.messaging.Outgoing

import io.smallrye.mutiny.Uni
import io.smallrye.mutiny.Multi

import io.quarkus.logging.Log

@ApplicationScoped
class TimedShortConsumer {
    @Incoming("shorts-in")
    fun batchedConsume(shorts: Multi<Short>) {
        shorts
            .onItem().call({ _ -> Uni.createFrom().item(Unit).onItem().delayIt().by(Duration.ofSeconds(1)) })
            .subscribe().with(
                { short: Short -> Log.info("Consumed " + short) }
            )
    }
}

But this doesn't work and errors for me with the following message:

 SRMSG00201: Error caught while processing a messa
ge in method playground.kafka.TimedShortConsumer#batchedConsume: java.lang.ClassCastException: class java.lang.Short
cannot be cast to class io.smallrye.mutiny.Multi (java.lang.Short is in module java.base of loader 'bootstrap'; io.smallrye.mutiny.Mu
lti is in unnamed module of loader io.quarkus.bootstrap.classloading.QuarkusClassLoader @1a38c59b)
        at mx.klar.quarkus.playground.kafka.TimedShortConsumer_SmallRyeMessagingInvoker_batchedConsume_988001ce33b80eb1f5aafeb45b383baf075982e3.invoke(Unknown Source)
        at io.smallrye.reactive.messaging.providers.AbstractMediator.invoke(AbstractMediator.java:160)
        at io.smallrye.reactive.messaging.providers.AbstractMediator.lambda$invokeOnMessageContext$13(AbstractMediator.java:168)
        at io.smallrye.reactive.messaging.providers.locals.LocalContextMetadata.lambda$invokeOnMessageContext$0(LocalContextMetadata.java:34)
        at io.smallrye.reactive.messaging.providers.locals.LocalContextMetadata.lambda$invokeOnMessageContext$1(LocalContextMetadata.java:53)
        at io.smallrye.reactive.messaging.providers.helpers.VertxContext.runOnContext(VertxContext.java:24)
        at io.smallrye.reactive.messaging.providers.locals.LocalContextMetadata.lambda$invokeOnMessageContext$2(LocalContextMetadata.java:51)
        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.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:74)
        at io.smallrye.mutiny.operators.multi.MultiConcatMapOp$MainSubscriber.innerOnSubscribe(MultiConcatMapOp.java:109)
        at io.smallrye.mutiny.operators.multi.MultiConcatMapOp$MainSubscriber$InnerSubscriber.onSubscribe(MultiConcatMapOp.java:297)
        at io.smallrye.mutiny.converters.uni.UniToMultiPublisher.subscribe(UniToMultiPublisher.java:25)
        at io.smallrye.mutiny.groups.MultiCreate$1.subscribe(MultiCreate.java:165)
        at io.smallrye.mutiny.operators.multi.MultiConcatMapOp$MainSubscriber.onItem(MultiConcatMapOp.java:119)
        at io.smallrye.mutiny.operators.multi.MultiMapOp$MapProcessor.onItem(MultiMapOp.java:50)
        at io.smallrye.mutiny.operators.multi.MultiOperatorProcessor.onItem(MultiOperatorProcessor.java:99)
        at io.smallrye.reactive.messaging.providers.locals.ContextOperator$ContextMulti$ContextProcessor.lambda$onItem$1(ContextOperator.java:71)
        at io.smallrye.reactive.messaging.providers.helpers.VertxContext.lambda$runOnContext$0(VertxContext.java:26)
        at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:276)
        at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:258)
        at io.vertx.core.impl.ContextInternal.lambda$runOnContext$0(ContextInternal.java:56)
        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:470)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
        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)

So instead I found on the Stack Overflow the following answer: https://stackoverflow.com/questions/70925787/reactive-messaging-classcastexception-cannot-be-cast-to-class-io-smallrye-muti and adapted my code accordingly. This works, but this is ugly. Why I have to resort to creating a new channel when I want to consume Multi<T>?

package playground.kafka

import jakarta.enterprise.context.ApplicationScoped

import java.time.Duration

import org.eclipse.microprofile.reactive.messaging.Incoming
import org.eclipse.microprofile.reactive.messaging.Outgoing

import io.smallrye.mutiny.Uni
import io.smallrye.mutiny.Multi

import io.quarkus.logging.Log

@ApplicationScoped
class TimedShortConsumer {
    @Incoming("shorts-in")
    @Outgoing("null")
    fun batchedConsume(shorts: Multi<Short>): Multi<Void> {
        return shorts
            .onItem().call({ _ -> Uni.createFrom().item(Unit).onItem().delayIt().by(Duration.ofSeconds(1)) })
            .onItem().invoke({ short: Short -> Log.info("Consumed " + short) })
            .onItem().ignore()
    }

    @Incoming("null")
    fun sink(nothing: Void) {}
}

[cescoffier]

Your method signature is not correct:

@Incoming("shorts-in")
    fun batchedConsume(shorts: Multi<Short>): Void {

should fix it.

See https://smallrye.io/smallrye-reactive-messaging/4.18.0/concepts/signatures/ for details.

[erplsf]

Probably I'm still not getting it, because removing @Outgoing signature and adding return type doesn't solve it - I still get the same error about casting when I rewrite my code as follows:

    @Incoming("shorts-in")
    fun batchedConsume(shorts: Multi<Short>): Multi<Void> {
        return shorts
            .onItem().call({ _ -> Uni.createFrom().item(Unit).onItem().delayIt().by(Duration.ofSeconds(1)) })
            .onItem().invoke({ short: Short -> Log.info("Consumed " + short) })
            .onItem().ignore()
    }

Here, if add the return type annotation I MUST return something from the method, and I cannot simply return Void.