fixes GroupBy overflow issue

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java

@@ -515,6 +515,7 @@ public K key() {
 		volatile boolean outputFused;
 
 		int produced;
+		boolean isFirstRequest = true;
 
 		UnicastGroupedFlux(K key,
 				Queue<V> queue,
@@ -572,7 +573,16 @@ void drainRegular(Subscriber<? super V> a) {
 				if (e != 0) {
 					GroupByMain<?, K, V> main = parent;
 					if (main != null) {
-						main.s.request(e);
+						if (this.isFirstRequest) {
+							this.isFirstRequest = false;
+							long toRequest = e - 1;
+
+							if (toRequest > 0) {
+								main.s.request(toRequest);
+							}
+						} else {
+							main.s.request(e);
+						}
 					}
 					if (r != Long.MAX_VALUE) {
 						REQUESTED.addAndGet(this, -e);
@@ -751,7 +761,16 @@ void tryReplenish() {
 				produced = 0;
 				GroupByMain<?, K, V> main = parent;
 				if (main != null) {
-					main.s.request(p);
+					if (this.isFirstRequest) {
+						this.isFirstRequest = false;
+						p--;
+
+						if (p > 0) {
+							main.s.request(p);
+						}
+					} else {
+						main.s.request(p);
+					}
 				}
 			}
 		}