Merge pull request #2003 from ozangunalp/group_by_over_request

Reproducer and fix for the groupBy operator upstream requests

Author: jponge

implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiGroupByOp.java

@@ -417,8 +417,6 @@ void drain() {
                             requested.addAndGet(-e);
                         }
                         parent.getUpstreamSubscription().request(e);
-                    } else {
-                        parent.getUpstreamSubscription().request(parent.prefetch);
                     }
                 }
 

implementation/src/test/java/io/smallrye/mutiny/operators/MultiGroupTest.java

@@ -14,10 +14,13 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.NoSuchElementException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Flow.Subscriber;
 import java.util.concurrent.Flow.Subscription;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
@@ -1037,4 +1040,47 @@ public void rejectGroupByBadPrefetch() {
         assertThrows(IllegalArgumentException.class,
                 () -> Multi.createFrom().range(1, 10).group().by(i -> i % 2, 0L));
     }
+
+    @Test
+    void testUpstreamRequestsNotBlownOutOfProportion() {
+        ExecutorService executor = Executors.newFixedThreadPool(10);
+        try {
+            AtomicLong requestCounter = new AtomicLong(0);
+            AtomicLong itemCounter = new AtomicLong(0);
+            AtomicReference<MultiEmitter<? super Integer>> e = new AtomicReference<>();
+
+            Multi.createFrom().<Integer>emitter(e::set)
+                    .onRequest().invoke(requestCounter::addAndGet)
+                    .group().by(i -> i / 10)
+                    .onItem().transformToMulti(g -> g.map(i -> g.key() + " : " + i)
+                            .emitOn(executor)
+                            .invoke(s -> {
+                                try {
+                                    Thread.sleep(100);
+                                    itemCounter.incrementAndGet();
+                                } catch (InterruptedException ex) {
+                                    throw new RuntimeException(ex);
+                                }
+                            }))
+                    .merge()
+                    .subscribe().withSubscriber(AssertSubscriber.create(Long.MAX_VALUE));
+
+            int itemCount = 100;
+            MultiEmitter<? super Integer> emitter = e.get();
+            new Thread(() -> {
+                int i = 0;
+                while (true) {
+                    if (emitter.requested() > 0) {
+                        emitter.emit(i);
+                        i++;
+                    }
+                }
+            }).start();
+
+            await().untilAsserted(() -> assertThat(itemCounter).hasValueGreaterThanOrEqualTo(itemCount));
+            assertThat(requestCounter.get()).isLessThan(10000L); // this should not blow up
+        } finally {
+            executor.shutdownNow();
+        }
+    }
 }