2.x: Distinct Operator - delegate null Collection down to onError (#4718)

Author: vanniktech

src/main/java/io/reactivex/internal/operators/flowable/FlowableDistinct.java

@@ -44,7 +44,7 @@ protected void subscribeActual(Subscriber<? super T> observer) {
         Collection<? super K> collection;
 
         try {
-            collection = collectionSupplier.call();
+            collection = ObjectHelper.requireNonNull(collectionSupplier.call(), "The collectionSupplier returned a null collection. Null values are generally not allowed in 2.x operators and sources.");
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
             EmptySubscription.error(ex, observer);

src/main/java/io/reactivex/internal/operators/observable/ObservableDistinct.java

@@ -43,7 +43,7 @@ protected void subscribeActual(Observer<? super T> observer) {
         Collection<? super K> collection;
 
         try {
-            collection = collectionSupplier.call();
+            collection = ObjectHelper.requireNonNull(collectionSupplier.call(), "The collectionSupplier returned a null collection. Null values are generally not allowed in 2.x operators and sources.");
         } catch (Throwable ex) {
             Exceptions.throwIfFatal(ex);
             EmptyDisposable.error(ex, observer);

src/test/java/io/reactivex/internal/operators/flowable/FlowableDistinctTest.java

@@ -212,6 +212,20 @@ public Collection<Object> call() throws Exception {
         .assertFailure(TestException.class);
     }
 
+    @Test
+    public void collectionSupplierIsNull() {
+        Flowable.just(1)
+        .distinct(Functions.identity(), new Callable<Collection<Object>>() {
+            @Override
+            public Collection<Object> call() throws Exception {
+                return null;
+            }
+        })
+        .test()
+        .assertFailure(NullPointerException.class)
+        .assertErrorMessage("The collectionSupplier returned a null collection. Null values are generally not allowed in 2.x operators and sources.");
+    }
+
     @Test
     public void badSource() {
         List<Throwable> errors = TestHelper.trackPluginErrors();

src/test/java/io/reactivex/internal/operators/observable/ObservableDistinctTest.java

@@ -213,6 +213,20 @@ public Collection<Object> call() throws Exception {
         .assertFailure(TestException.class);
     }
 
+    @Test
+    public void collectionSupplierIsNull() {
+        Observable.just(1)
+        .distinct(Functions.identity(), new Callable<Collection<Object>>() {
+            @Override
+            public Collection<Object> call() throws Exception {
+                return null;
+            }
+        })
+        .test()
+        .assertFailure(NullPointerException.class)
+        .assertErrorMessage("The collectionSupplier returned a null collection. Null values are generally not allowed in 2.x operators and sources.");
+    }
+
     @Test
     public void badSource() {
         List<Throwable> errors = TestHelper.trackPluginErrors();