Ensure no NPE is thrown with SerializedFluxSink (#1096)

When the FluxSink is an instance of SerializedFluxSink and the results
contains a null value, a NullPointerException is thrown and the whole process
stops. This is because SerializedFluxSink.next performs a call to
Object.requireNonNull().

To fix this issue we filter out null values from the results.

JAVA-4908

Author: imclem

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/BatchCursorFlux.java

@@ -24,6 +24,7 @@
 import reactor.core.publisher.Mono;
 import reactor.util.context.Context;
 
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -95,7 +96,10 @@ private void recurseCursor(){
                         })
                         .doOnSuccess(results -> {
                             if (results != null) {
-                                results.forEach(sink::next);
+                                results
+                                        .stream()
+                                        .filter(Objects::nonNull)
+                                        .forEach(sink::next);
                                 calculateDemand(-results.size());
                             }
                             if (batchCursor.isClosed()) {

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/internal/BatchCursorFluxTest.java

@@ -46,6 +46,8 @@
 import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
 
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
@@ -328,6 +330,32 @@ public void testBatchCursorDoesNotDropAnError() {
         }
     }
 
+    @Test
+    @DisplayName("Ensure no NPE is thrown on null in result set")
+    public void testNoNPEOnNull() {
+        try {
+            AtomicBoolean errorDropped = new AtomicBoolean();
+            Hooks.onErrorDropped(t -> errorDropped.set(true));
+
+            Document doc = new Document("x", null);
+            Document doc2 = new Document("x", "hello");
+
+            Mono.from(collection.insertMany(Arrays.asList(doc, doc2))).block();
+
+            TestSubscriber<String> subscriber = new TestSubscriber<>();
+
+            collection.distinct("x", String.class).subscribe(subscriber);
+
+            subscriber.requestMore(1);
+
+            subscriber.assertReceivedOnNext(Arrays.asList("hello"));
+
+            assertFalse(errorDropped.get());
+        } finally {
+            Hooks.resetOnErrorDropped();
+        }
+    }
+
     @Test
     @DisplayName("Ensure BatchCursor reports cursor errors")
     @SuppressWarnings("OptionalGetWithoutIsPresent")