Quarkus Reactive MongoDB Client return null when watch a database

[ruget]

Describe the bug

This issue occurred when we register to a MongoDB Change Stream with options.

mongoClient.getDatabase(mongodbDatabaseName)
   .watch(new ChangeStreamOptions().fullDocument(FullDocument.UPDATE_LOOKUP);

This code return an unexpected null

Seams to be releated to this part of the code:

quarkus/extensions/mongodb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoDatabaseImpl.java

Line 210 in ee2c00b

public Multi<ChangeStreamDocument<Document>> watch(ChangeStreamOptions options) {

Expected behavior

Multi<ChangeStreamDocument<Document>> is returned instead of null to listen MongoDB change on all database collections available.

Actual behavior

null is returned instead of the expected Multi.

How to Reproduce?

1. Add the extension quarkus-mongodb-client in the pom.xml

    <dependency>
            <groupId>io.quarkus</groupId>
            <artifactId>quarkus-mongodb-client</artifactId>
        </dependency>

2. Call the following method

mongoClient.getDatabase(mongodbDatabaseName) .watch(new ChangeStreamOptions().fullDocument(FullDocument.UPDATE_LOOKUP)).subscribe().with(dcsd -> {})

3. java.lang.NullPointerException is thrown

ERROR [com.sam.mdm.ser.StreamService] (vert.x-eventloop-thread-3) Stream flow issue: java.lang.NullPointerException: The mapper returned `null`
       at io.smallrye.mutiny.operators.uni.UniOnItemTransformToMulti$FlatMapPublisherSubscriber.onItem(UniOnItemTransformToMulti.java:132)
       at io.smallrye.mutiny.operators.uni.UniOnItemConsume$UniOnItemComsumeProcessor.onItem(UniOnItemConsume.java:43)
       at io.smallrye.mutiny.operators.uni.builders.UniCreateFromPublisher$PublisherSubscriber.onNext(UniCreateFromPublisher.java:73)
       at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)
       at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:85)
       at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)
       at io.smallrye.mutiny.subscription.SerializedSubscriber.onItem(SerializedSubscriber.java:75)
       at io.smallrye.mutiny.operators.multi.MultiRetryWhenOp$RetryWhenOperator.onItem(MultiRetryWhenOp.java:113)
       at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)
       at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.onItem(UniToMultiPublisher.java:93)
       at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:60)
       at io.smallrye.mutiny.operators.uni.UniOnItemTransform$UniOnItemTransformProcessor.onItem(UniOnItemTransform.java:43)
       at io.smallrye.mutiny.operators.uni.builders.UniCreateFromPublisher$PublisherSubscriber.onNext(UniCreateFromPublisher.java:73)
       at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)
       at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:85)
       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.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:264)
       at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:246)
       at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:43)
       at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
       at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
       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:834)

Output of uname -a or ver

No response

Output of java -version

java version "11.0.1" 2018-10-16 LTS

GraalVM version (if different from Java)

No response

Quarkus version or git rev

2.13.3.Final

Build tool (ie. output of mvnw --version or gradlew --version)

Apache Maven 3.8.6 (84538c9988a25aec085021c365c560670ad80f63)
Maven home: C:\Users\jruget1.m2\wrapper\dists\apache-maven-3.8.6-bin\67568434\apache-maven-3.8.6
Java version: 11.0.1, vendor: Oracle Corporation, runtime: C:\Tools\jdk\current
Default locale: fr_FR, platform encoding: Cp1252
OS name: "windows 10", version: "10.0", arch: "amd64", family: "windows"

Additional information

No response

[ruget]

Started a PR proposal #28828

[quarkus-bot]

/cc @evanchooly, @loicmathieu

[loicmathieu]

Fixes by #28828