Unable to initialize mediator without an Reactive Streams Operators in the classpath

[matzew]

Working on an application that reads Apache Kafka messages:

package net.wessendorf.knative;

import io.smallrye.reactive.messaging.kafka.KafkaMessage;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.enterprise.context.ApplicationScoped;
import java.io.IOException;
import java.util.concurrent.CompletionStage;

@ApplicationScoped
public class KafkaEventConsumer{

    private static final Logger LOG = LoggerFactory.getLogger(KafkaEventConsumer.class);

    @Incoming("meters")
    public CompletionStage<Void> onMessage(KafkaMessage<String, String> message) throws IOException {

        LOG.info("Kafka message with value = {} arrived", message.getPayload());
        return message.ack();
    }

}

and in DEV MODE I am able to get this running and I see messages flying in.

However, when doing native image build, and starting the runner, I do get this error:

2019-03-14 18:08:43,122 ERROR [io.sma.rea.mes.ext.MediatorManager] (main) Unable to initialize mediator: net.wessendorf.knative.KafkaEventConsumer#onMessage: java.lang.IllegalStateException: No ReactiveStreamsFactory implementation found!
	at org.eclipse.microprofile.reactive.streams.operators.spi.ReactiveStreamsFactoryResolver.instance(ReactiveStreamsFactoryResolver.java:64)
	at org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams.builder(ReactiveStreams.java:152)
	at io.smallrye.reactive.messaging.SubscriberMediator.processMethodReturningACompletionStage(SubscriberMediator.java:147)
	at io.smallrye.reactive.messaging.SubscriberMediator.initialize(SubscriberMediator.java:53)
	at io.smallrye.reactive.messaging.extension.MediatorManager.lambda$null$5(MediatorManager.java:143)
	at java.util.ArrayList.forEach(ArrayList.java:1257)
	at io.smallrye.reactive.messaging.extension.MediatorManager.lambda$initializeAndRun$6(MediatorManager.java:125)
	at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656)
	at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:669)
	at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:1997)
	at io.smallrye.reactive.messaging.extension.MediatorManager.initializeAndRun(MediatorManager.java:120)
	at io.smallrye.reactive.messaging.extension.MediatorManager_ClientProxy.initializeAndRun(Unknown Source)
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:20)
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_4e8937813d9e8faff65c3c07f88fa96615b70e70.notify(Unknown Source)
	at io.quarkus.arc.EventImpl$Notifier.notify(EventImpl.java:226)
	at io.quarkus.arc.EventImpl.fire(EventImpl.java:85)
	at io.quarkus.arc.runtime.LifecycleEventRunner.fireStartupEvent(LifecycleEventRunner.java:39)
	at io.quarkus.arc.runtime.ArcDeploymentTemplate.handleLifecycleEvents(ArcDeploymentTemplate.java:106)
	at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent4.deploy(Unknown Source)
	at io.quarkus.runner.ApplicationImpl1.doStart(Unknown Source)
	at io.quarkus.runtime.Application.start(Application.java:93)
	at io.quarkus.runtime.Application.run(Application.java:198)
	at io.quarkus.runner.GeneratedMain.main(Unknown Source)

2019-03-14 18:08:43,122 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Connecting mediators
2019-03-14 18:08:43,122 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Attempt to resolve net.wessendorf.knative.KafkaEventConsumer#onMessage
2019-03-14 18:08:43,123 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Connecting net.wessendorf.knative.KafkaEventConsumer#onMessage to `meters` (io.reactivex.internal.operators.flowable.FlowableMap@7f3cfc2888a8)
java.lang.RuntimeException: java.lang.NullPointerException
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:24)
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_4e8937813d9e8faff65c3c07f88fa96615b70e70.notify(Unknown Source)
	at io.quarkus.arc.EventImpl$Notifier.notify(EventImpl.java:226)
	at io.quarkus.arc.EventImpl.fire(EventImpl.java:85)
	at io.quarkus.arc.runtime.LifecycleEventRunner.fireStartupEvent(LifecycleEventRunner.java:39)
	at io.quarkus.arc.runtime.ArcDeploymentTemplate.handleLifecycleEvents(ArcDeploymentTemplate.java:106)
	at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent4.deploy(Unknown Source)
	at io.quarkus.runner.ApplicationImpl1.doStart(Unknown Source)
	at io.quarkus.runtime.Application.start(Application.java:93)
	at io.quarkus.runtime.Application.run(Application.java:198)
	at io.quarkus.runner.GeneratedMain.main(Unknown Source)
Caused by: java.lang.NullPointerException
	at io.smallrye.reactive.messaging.SubscriberMediator$1.onSubscribe(SubscriberMediator.java:92)
	at io.reactivex.internal.subscribers.StrictSubscriber.onSubscribe(StrictSubscriber.java:85)
	at io.reactivex.internal.subscribers.BasicFuseableSubscriber.onSubscribe(BasicFuseableSubscriber.java:67)
	at io.reactivex.internal.operators.flowable.FlowableRetryWhen.subscribeActual(FlowableRetryWhen.java:58)
	at io.reactivex.Flowable.subscribe(Flowable.java:14805)
	at io.reactivex.internal.operators.flowable.FlowableMap.subscribeActual(FlowableMap.java:37)
	at io.reactivex.Flowable.subscribe(Flowable.java:14805)
	at io.reactivex.Flowable.subscribe(Flowable.java:14755)
	at io.smallrye.reactive.messaging.SubscriberMediator.run(SubscriberMediator.java:113)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
	at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
	at io.smallrye.reactive.messaging.extension.MediatorManager.weaving(MediatorManager.java:229)
	at io.smallrye.reactive.messaging.extension.MediatorManager.lambda$initializeAndRun$6(MediatorManager.java:169)
	at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656)
	at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:669)
	at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:1997)
	at io.smallrye.reactive.messaging.extension.MediatorManager.initializeAndRun(MediatorManager.java:120)
	at io.smallrye.reactive.messaging.extension.MediatorManager_ClientProxy.initializeAndRun(Unknown Source)
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:20)
	... 10 more
2019-03-14 18:08:43,123 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Cancel subscriptions
Exception in thread "main" java.lang.RuntimeException: Failed to start quarkus
	at io.quarkus.runner.ApplicationImpl1.doStart(Unknown Source)
	at io.quarkus.runtime.Application.start(Application.java:93)
	at io.quarkus.runtime.Application.run(Application.java:198)
	at io.quarkus.runner.GeneratedMain.main(Unknown Source)
Caused by: java.lang.RuntimeException: java.lang.NullPointerException
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:24)
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_4e8937813d9e8faff65c3c07f88fa96615b70e70.notify(Unknown Source)
	at io.quarkus.arc.EventImpl$Notifier.notify(EventImpl.java:226)
	at io.quarkus.arc.EventImpl.fire(EventImpl.java:85)
	at io.quarkus.arc.runtime.LifecycleEventRunner.fireStartupEvent(LifecycleEventRunner.java:39)
	at io.quarkus.arc.runtime.ArcDeploymentTemplate.handleLifecycleEvents(ArcDeploymentTemplate.java:106)
	at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent4.deploy(Unknown Source)
	... 4 more
Caused by: java.lang.NullPointerException
	at io.smallrye.reactive.messaging.SubscriberMediator$1.onSubscribe(SubscriberMediator.java:92)
	at io.reactivex.internal.subscribers.StrictSubscriber.onSubscribe(StrictSubscriber.java:85)
	at io.reactivex.internal.subscribers.BasicFuseableSubscriber.onSubscribe(BasicFuseableSubscriber.java:67)
	at io.reactivex.internal.operators.flowable.FlowableRetryWhen.subscribeActual(FlowableRetryWhen.java:58)
	at io.reactivex.Flowable.subscribe(Flowable.java:14805)
	at io.reactivex.internal.operators.flowable.FlowableMap.subscribeActual(FlowableMap.java:37)
	at io.reactivex.Flowable.subscribe(Flowable.java:14805)
	at io.reactivex.Flowable.subscribe(Flowable.java:14755)
	at io.smallrye.reactive.messaging.SubscriberMediator.run(SubscriberMediator.java:113)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
	at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
	at io.smallrye.reactive.messaging.extension.MediatorManager.weaving(MediatorManager.java:229)
	at io.smallrye.reactive.messaging.extension.MediatorManager.lambda$initializeAndRun$6(MediatorManager.java:169)
	at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656)
	at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:669)
	at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:1997)
	at io.smallrye.reactive.messaging.extension.MediatorManager.initializeAndRun(MediatorManager.java:120)
	at io.smallrye.reactive.messaging.extension.MediatorManager_ClientProxy.initializeAndRun(Unknown Source)
	at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:20)
	... 10 more

[matzew]

and my dependencies are:

  <dependencies>
    <dependency>
      <groupId>org.glassfish</groupId>
      <artifactId>javax.json</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-smallrye-reactive-messaging-kafka</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-smallrye-reactive-messaging</artifactId>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-core</artifactId>
    </dependency>
  </dependencies>

[gsmet]

I suspect we would need to register the ReactiveStreamsFactory implementations for reflection. Do you have a specific one in your app?

If so, could you mark your implementation with @RegisterForReflection and see if it fixes your issue?

[matzew]

only that one class I have

On Thu 14. Mar 2019 at 20:37, Guillaume Smet ***@***.***> wrote: I suspect we would need to register the ReactiveStreamsFactory implementations for reflection. Do you have a specific one in your app? If so, could you mark your implementation with @RegisterForReflection and see if it fixes your issue? — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#1495 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAJnzhdJhMxO-e2_KgVsWxdqYbSSdmvmks5vWqUGgaJpZM4b0yCi> .

-- Sent from Gmail Mobile

[gsmet]

OK, looks like a bug then. Thanks for reporting.

[matzew]

@gsmet FWIW, here is my code: https://github.com/matzew/quarfka-source

[cescoffier]

I will have a look. Thanks!

@mkouba did you already see such kind of issue?

[mkouba]

So the problem is that there is no ReactiveStreamsFactory service provider registered. This SPI is implemented in smallrye-reactive-streams-operators. It works in JDK mode because smallrye-reactive-streams-operators is a transitive dependency of quarkus-smallrye-reactive-messaging-kafka. However, in order to make it work on substrate we need to register a ServiceProviderBuildItem -> implemented in io.quarkus:quarkus-smallrye-reactive-streams-operators.

WORKAROUND: add the io.quarkus:quarkus-smallrye-reactive-streams-operators dependency to the application pom.xml. @matzew could you verify this pls?

PROPOSED FIX: add the io.quarkus:quarkus-smallrye-reactive-streams-operators dependency to the quarkus-smallrye-reactive-messaging deployment module pom.xml. @cescoffier does it sound reasonable?

[matzew]

yay~

2019-03-18 09:36:11,204 WARN  [org.apa.kaf.cli.con.ConsumerConfig] (main) The configuration 'topic' was supplied but isn't a known config.
2019-03-18 09:36:11,204 WARN  [org.apa.kaf.cli.con.ConsumerConfig] (main) The configuration 'type' was supplied but isn't a known config.
2019-03-18 09:36:11,204 INFO  [io.sma.rea.mes.imp.ConfiguredStreamFactory] (main) Publishers created during initialization: [meters]
2019-03-18 09:36:11,204 INFO  [io.sma.rea.mes.imp.ConfiguredStreamFactory] (main) Subscribers created during initialization: []
2019-03-18 09:36:11,204 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Initializing mediators
2019-03-18 09:36:11,204 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Connecting mediators
2019-03-18 09:36:11,204 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Attempt to resolve net.wessendorf.knative.KafkaEventConsumer#onMessage
2019-03-18 09:36:11,204 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Connecting net.wessendorf.knative.KafkaEventConsumer#onMessage to `meters` (io.reactivex.internal.operators.flowable.FlowableMap@7fac1698a530)
2019-03-18 09:36:11,204 INFO  [io.quarkus] (main) Quarkus 0.11.0 started in 0.008s. 

[matzew]

great suggestion, thanks @mkouba

now, I think @cescoffier suggested that already, but I understood it wrong, and added this instead:

      <artifactId>quarkus-smallrye-reactive-messaging</artifactId>

[cescoffier]

@mkouba I agree with the proposed fix, opening a PR in a few minutes.