3.x: Fix groupBy group emission in some cases (#6664)

* 3.x: Fix Observable.groupBy group emission in some cases

* Test non-fresh group with value selector failure

Author: akarnokd

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java

@@ -168,6 +168,10 @@ public void onNext(T t) {
             } catch (Throwable ex) {
                 Exceptions.throwIfFatal(ex);
                 upstream.cancel();
+                if (newGroup) {
+                    q.offer(group);
+                    drain();
+                }
                 onError(ex);
                 return;
             }

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableGroupBy.java

@@ -97,6 +97,7 @@ public void onNext(T t) {
 
             Object mapKey = key != null ? key : NULL_KEY;
             GroupedUnicast<K, V> group = groups.get(mapKey);
+            boolean newGroup = false;
             if (group == null) {
                 // if the main has been cancelled, stop creating groups
                 // and skip this value
@@ -109,12 +110,7 @@ public void onNext(T t) {
 
                 getAndIncrement();
 
-                downstream.onNext(group);
-
-                if (group.state.tryAbandon()) {
-                    cancel(key);
-                    group.onComplete();
-                }
+                newGroup = true;
             }
 
             V v;
@@ -123,11 +119,23 @@ public void onNext(T t) {
             } catch (Throwable e) {
                 Exceptions.throwIfFatal(e);
                 upstream.dispose();
+                if (newGroup) {
+                    downstream.onNext(group);
+                }
                 onError(e);
                 return;
             }
 
             group.onNext(v);
+
+            if (newGroup) {
+                downstream.onNext(group);
+
+                if (group.state.tryAbandon()) {
+                    cancel(key);
+                    group.onComplete();
+                }
+            }
         }
 
         @Override

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupByTest.java

@@ -2275,4 +2275,61 @@ public void accept(GroupedFlowable<Integer, Integer> v) throws Throwable {
         .assertNoErrors()
         .assertComplete();
     }
+
+    @Test
+    public void newGroupValueSelectorFails() {
+        TestSubscriber<Object> ts1 = new TestSubscriber<Object>();
+        final TestSubscriber<Object> ts2 = new TestSubscriber<Object>();
+
+        Flowable.just(1)
+        .groupBy(Functions.<Integer>identity(), new Function<Integer, Object>() {
+            @Override
+            public Object apply(Integer v) throws Throwable {
+                throw new TestException();
+            }
+        })
+        .doOnNext(new Consumer<GroupedFlowable<Integer, Object>>() {
+            @Override
+            public void accept(GroupedFlowable<Integer, Object> g) throws Throwable {
+                g.subscribe(ts2);
+            }
+        })
+        .subscribe(ts1);
+
+        ts1.assertValueCount(1)
+        .assertError(TestException.class)
+        .assertNotComplete();
+
+        ts2.assertFailure(TestException.class);
+    }
+
+    @Test
+    public void existingGroupValueSelectorFails() {
+        TestSubscriber<Object> ts1 = new TestSubscriber<Object>();
+        final TestSubscriber<Object> ts2 = new TestSubscriber<Object>();
+
+        Flowable.just(1, 2)
+        .groupBy(Functions.justFunction(1), new Function<Integer, Object>() {
+            @Override
+            public Object apply(Integer v) throws Throwable {
+                if (v == 2) {
+                    throw new TestException();
+                }
+                return v;
+            }
+        })
+        .doOnNext(new Consumer<GroupedFlowable<Integer, Object>>() {
+            @Override
+            public void accept(GroupedFlowable<Integer, Object> g) throws Throwable {
+                g.subscribe(ts2);
+            }
+        })
+        .subscribe(ts1);
+
+        ts1.assertValueCount(1)
+        .assertError(TestException.class)
+        .assertNotComplete();
+
+        ts2.assertFailure(TestException.class, 1);
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableGroupByTest.java

@@ -1615,4 +1615,61 @@ public void accept(GroupedObservable<Integer, Integer> v) throws Throwable {
         .assertNoErrors()
         .assertComplete();
     }
+
+    @Test
+    public void newGroupValueSelectorFails() {
+        TestObserver<Object> to1 = new TestObserver<Object>();
+        final TestObserver<Object> to2 = new TestObserver<Object>();
+
+        Observable.just(1)
+        .groupBy(Functions.<Integer>identity(), new Function<Integer, Object>() {
+            @Override
+            public Object apply(Integer v) throws Throwable {
+                throw new TestException();
+            }
+        })
+        .doOnNext(new Consumer<GroupedObservable<Integer, Object>>() {
+            @Override
+            public void accept(GroupedObservable<Integer, Object> g) throws Throwable {
+                g.subscribe(to2);
+            }
+        })
+        .subscribe(to1);
+
+        to1.assertValueCount(1)
+        .assertError(TestException.class)
+        .assertNotComplete();
+
+        to2.assertFailure(TestException.class);
+    }
+
+    @Test
+    public void existingGroupValueSelectorFails() {
+        TestObserver<Object> to1 = new TestObserver<Object>();
+        final TestObserver<Object> to2 = new TestObserver<Object>();
+
+        Observable.just(1, 2)
+        .groupBy(Functions.justFunction(1), new Function<Integer, Object>() {
+            @Override
+            public Object apply(Integer v) throws Throwable {
+                if (v == 2) {
+                    throw new TestException();
+                }
+                return v;
+            }
+        })
+        .doOnNext(new Consumer<GroupedObservable<Integer, Object>>() {
+            @Override
+            public void accept(GroupedObservable<Integer, Object> g) throws Throwable {
+                g.subscribe(to2);
+            }
+        })
+        .subscribe(to1);
+
+        to1.assertValueCount(1)
+        .assertError(TestException.class)
+        .assertNotComplete();
+
+        to2.assertFailure(TestException.class, 1);
+    }
 }