2.x: Fix parallel() on grouped flowable not replenishing properly (#6720)

Author: akarnokd

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

@@ -709,17 +709,25 @@ public T poll() {
                 produced++;
                 return v;
             }
-            int p = produced;
-            if (p != 0) {
-                produced = 0;
-                parent.upstream.request(p);
-            }
+            tryReplenish();
             return null;
         }
 
         @Override
         public boolean isEmpty() {
-            return queue.isEmpty();
+            if (queue.isEmpty()) {
+                tryReplenish();
+                return true;
+            }
+            return false;
+        }
+
+        void tryReplenish() {
+            int p = produced;
+            if (p != 0) {
+                produced = 0;
+                parent.upstream.request(p);
+            }
         }
 
         @Override

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

@@ -2286,4 +2286,34 @@ public void run() {
             }
         }
     }
+
+    @Test
+    public void fusedParallelGroupProcessing() {
+        Flowable.range(0, 500000)
+        .subscribeOn(Schedulers.single())
+        .groupBy(new Function<Integer, Integer>() {
+            @Override
+            public Integer apply(Integer i) {
+                return i % 2;
+            }
+        })
+        .flatMap(new Function<GroupedFlowable<Integer, Integer>, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(GroupedFlowable<Integer, Integer> g) {
+                return g.getKey() == 0
+                    ? g
+                        .parallel()
+                        .runOn(Schedulers.computation())
+                        .map(Functions.<Integer>identity())
+                        .sequential()
+                    : g.map(Functions.<Integer>identity()) // no need to use hide
+                ;
+            }
+        })
+        .test()
+        .awaitDone(20, TimeUnit.SECONDS)
+        .assertValueCount(500000)
+        .assertComplete()
+        .assertNoErrors();
+    }
 }