3.x: Fix groupBy not canceling upstream due to group abandonment

.travis.yml

@@ -13,7 +13,7 @@ script: gradle/buildViaTravis.sh
 
 # Code coverage
 after_success:
-  - bash <(curl -s https://codecov.io/bash)
+  - bash <(curl -s --retry 10 https://codecov.io/bash)
   - bash gradle/push_javadoc.sh
 
 # cache between builds

src/main/java/io/reactivex/rxjava3/core/Flowable.java

@@ -10414,6 +10414,11 @@ public final Disposable forEachWhile(final Predicate<? super T> onNext, final Co
      * {@link #flatMap(Function, int)} or {@link #concatMapEager(Function, int, int)} and overriding the default maximum concurrency
      * value to be greater or equal to the expected number of groups, possibly using
      * {@code Integer.MAX_VALUE} if the number of expected groups is unknown.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
      *
      * <dl>
      *  <dt><b>Backpressure:</b></dt>
@@ -10462,6 +10467,12 @@ public final <K> Flowable<GroupedFlowable<K, T>> groupBy(Function<? super T, ? e
      * {@link #flatMap(Function, int)} or {@link #concatMapEager(Function, int, int)} and overriding the default maximum concurrency
      * value to be greater or equal to the expected number of groups, possibly using
      * {@code Integer.MAX_VALUE} if the number of expected groups is unknown.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
+     *
      * <dl>
      *  <dt><b>Backpressure:</b></dt>
      *  <dd>Both the returned and its inner {@code Publisher}s honor backpressure and the source {@code Publisher}
@@ -10512,6 +10523,11 @@ public final <K> Flowable<GroupedFlowable<K, T>> groupBy(Function<? super T, ? e
      * {@link #flatMap(Function, int)} or {@link #concatMapEager(Function, int, int)} and overriding the default maximum concurrency
      * value to be greater or equal to the expected number of groups, possibly using
      * {@code Integer.MAX_VALUE} if the number of expected groups is unknown.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
      *
      * <dl>
      *  <dt><b>Backpressure:</b></dt>
@@ -10565,6 +10581,11 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
      * {@link #flatMap(Function, int)} or {@link #concatMapEager(Function, int, int)} and overriding the default maximum concurrency
      * value to be greater or equal to the expected number of groups, possibly using
      * {@code Integer.MAX_VALUE} if the number of expected groups is unknown.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
      *
      * <dl>
      *  <dt><b>Backpressure:</b></dt>
@@ -10621,6 +10642,11 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
      * {@link #flatMap(Function, int)} or {@link #concatMapEager(Function, int, int)} and overriding the default maximum concurrency
      * value to be greater or equal to the expected number of groups, possibly using
      * {@code Integer.MAX_VALUE} if the number of expected groups is unknown.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
      *
      * <dl>
      *  <dt><b>Backpressure:</b></dt>
@@ -10726,6 +10752,11 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
      * {@link #flatMap(Function, int)} or {@link #concatMapEager(Function, int, int)} and overriding the default maximum concurrency
      * value to be greater or equal to the expected number of groups, possibly using
      * {@code Integer.MAX_VALUE} if the number of expected groups is unknown.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
      *
      * <dl>
      *  <dt><b>Backpressure:</b></dt>

src/main/java/io/reactivex/rxjava3/core/Observable.java

@@ -9067,6 +9067,12 @@ public final Disposable forEachWhile(final Predicate<? super T> onNext, Consumer
      * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those
      * {@code GroupedObservableSource}s that do not concern you. Instead, you can signal to them that they may
      * discard their buffers by applying an operator like {@link #ignoreElements} to them.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
+     *
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code groupBy} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -9101,6 +9107,12 @@ public final <K> Observable<GroupedObservable<K, T>> groupBy(Function<? super T,
      * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those
      * {@code GroupedObservableSource}s that do not concern you. Instead, you can signal to them that they may
      * discard their buffers by applying an operator like {@link #ignoreElements} to them.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
+     *
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code groupBy} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -9138,6 +9150,12 @@ public final <K> Observable<GroupedObservable<K, T>> groupBy(Function<? super T,
      * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those
      * {@code GroupedObservableSource}s that do not concern you. Instead, you can signal to them that they may
      * discard their buffers by applying an operator like {@link #ignoreElements} to them.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
+     *
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code groupBy} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -9176,6 +9194,12 @@ public final <K, V> Observable<GroupedObservable<K, V>> groupBy(Function<? super
      * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those
      * {@code GroupedObservableSource}s that do not concern you. Instead, you can signal to them that they may
      * discard their buffers by applying an operator like {@link #ignoreElements} to them.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
+     *
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code groupBy} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -9217,6 +9241,12 @@ public final <K, V> Observable<GroupedObservable<K, V>> groupBy(Function<? super
      * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those
      * {@code GroupedObservableSource}s that do not concern you. Instead, you can signal to them that they may
      * discard their buffers by applying an operator like {@link #ignoreElements} to them.
+     * <p>
+     * Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
+     * so-called group abandonment where a group will only contain one element and the group will be
+     * re-created over and over as new upstream items trigger a new group. The behavior is
+     * a tradeoff between no-dataloss, upstream cancellation and excessive group creation.
+     *
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code groupBy} does not operate by default on a particular {@link Scheduler}.</dd>

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

@@ -179,6 +179,13 @@ public void onNext(T t) {
             if (newGroup) {
                 q.offer(group);
                 drain();
+
+                if (group.state.tryAbandon()) {
+                    cancel(key);
+                    group.onComplete();
+
+                    upstream.request(1);
+                }
             }
         }
 
@@ -489,12 +496,17 @@ static final class State<T, K> extends BasicIntQueueSubscription<T> implements P
 
         final AtomicReference<Subscriber<? super T>> actual = new AtomicReference<Subscriber<? super T>>();
 
-        final AtomicBoolean once = new AtomicBoolean();
-
         boolean outputFused;
 
         int produced;
 
+        final AtomicInteger once = new AtomicInteger();
+
+        static final int FRESH = 0;
+        static final int HAS_SUBSCRIBER = 1;
+        static final int ABANDONED = 2;
+        static final int ABANDONED_HAS_SUBSCRIBER = ABANDONED | HAS_SUBSCRIBER;
+
         State(int bufferSize, GroupBySubscriber<?, K, T> parent, K key, boolean delayError) {
             this.queue = new SpscLinkedArrayQueue<T>(bufferSize);
             this.parent = parent;
@@ -513,19 +525,30 @@ public void request(long n) {
         @Override
         public void cancel() {
             if (cancelled.compareAndSet(false, true)) {
-                parent.cancel(key);
+                cancelParent();
             }
         }
 
         @Override
-        public void subscribe(Subscriber<? super T> s) {
-            if (once.compareAndSet(false, true)) {
-                s.onSubscribe(this);
-                actual.lazySet(s);
-                drain();
-            } else {
-                EmptySubscription.error(new IllegalStateException("Only one Subscriber allowed!"), s);
+        public void subscribe(Subscriber<? super T> subscriber) {
+            for (;;) {
+                int s = once.get();
+                if ((s & HAS_SUBSCRIBER) != 0) {
+                    break;
+                }
+                int u = s | HAS_SUBSCRIBER;
+                if (once.compareAndSet(s, u)) {
+                    subscriber.onSubscribe(this);
+                    actual.lazySet(subscriber);
+                    if (cancelled.get()) {
+                        actual.lazySet(null);
+                    } else {
+                        drain();
+                    }
+                    return;
+                }
             }
+            EmptySubscription.error(new IllegalStateException("Only one Subscriber allowed!"), subscriber);
         }
 
         public void onNext(T t) {
@@ -544,6 +567,16 @@ public void onComplete() {
             drain();
         }
 
+        void cancelParent() {
+            if ((once.get() & ABANDONED) == 0) {
+                parent.cancel(key);
+            }
+        }
+
+        boolean tryAbandon() {
+            return once.get() == FRESH && once.compareAndSet(FRESH, ABANDONED);
+        }
+
         void drain() {
             if (getAndIncrement() != 0) {
                 return;
@@ -640,7 +673,9 @@ void drainNormal() {
                         if (r != Long.MAX_VALUE) {
                             requested.addAndGet(-e);
                         }
-                        parent.upstream.request(e);
+                        if ((once.get() & ABANDONED) == 0) {
+                            parent.upstream.request(e);
+                        }
                     }
                 }
 
@@ -708,7 +743,9 @@ public T poll() {
             int p = produced;
             if (p != 0) {
                 produced = 0;
-                parent.upstream.request(p);
+                if ((once.get() & ABANDONED) == 0) {
+                    parent.upstream.request(p);
+                }
             }
             return null;
         }

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

@@ -110,6 +110,11 @@ public void onNext(T t) {
                 getAndIncrement();
 
                 downstream.onNext(group);
+
+                if (group.state.tryAbandon()) {
+                    cancel(key);
+                    group.onComplete();
+                }
             }
 
             V v;
@@ -151,7 +156,7 @@ public void onComplete() {
 
         @Override
         public void dispose() {
-            // cancelling the main source means we don't want any more groups
+            // canceling the main source means we don't want any more groups
             // but running groups still require new values
             if (cancelled.compareAndSet(false, true)) {
                 if (decrementAndGet() == 0) {
@@ -220,10 +225,15 @@ static final class State<T, K> extends AtomicInteger implements Disposable, Obse
 
         final AtomicBoolean cancelled = new AtomicBoolean();
 
-        final AtomicBoolean once = new AtomicBoolean();
-
         final AtomicReference<Observer<? super T>> actual = new AtomicReference<Observer<? super T>>();
 
+        final AtomicInteger once = new AtomicInteger();
+
+        static final int FRESH = 0;
+        static final int HAS_SUBSCRIBER = 1;
+        static final int ABANDONED = 2;
+        static final int ABANDONED_HAS_SUBSCRIBER = ABANDONED | HAS_SUBSCRIBER;
+
         State(int bufferSize, GroupByObserver<?, K, T> parent, K key, boolean delayError) {
             this.queue = new SpscLinkedArrayQueue<T>(bufferSize);
             this.parent = parent;
@@ -236,7 +246,7 @@ public void dispose() {
             if (cancelled.compareAndSet(false, true)) {
                 if (getAndIncrement() == 0) {
                     actual.lazySet(null);
-                    parent.cancel(key);
+                    cancelParent();
                 }
             }
         }
@@ -248,17 +258,24 @@ public boolean isDisposed() {
 
         @Override
         public void subscribe(Observer<? super T> observer) {
-            if (once.compareAndSet(false, true)) {
-                observer.onSubscribe(this);
-                actual.lazySet(observer);
-                if (cancelled.get()) {
-                    actual.lazySet(null);
-                } else {
-                    drain();
+            for (;;) {
+                int s = once.get();
+                if ((s & HAS_SUBSCRIBER) != 0) {
+                    break;
+                }
+                int u = s | HAS_SUBSCRIBER;
+                if (once.compareAndSet(s, u)) {
+                    observer.onSubscribe(this);
+                    actual.lazySet(observer);
+                    if (cancelled.get()) {
+                        actual.lazySet(null);
+                    } else {
+                        drain();
+                    }
+                    return;
                 }
-            } else {
-                EmptyDisposable.error(new IllegalStateException("Only one Observer allowed!"), observer);
             }
+            EmptyDisposable.error(new IllegalStateException("Only one Observer allowed!"), observer);
         }
 
         public void onNext(T t) {
@@ -315,11 +332,21 @@ void drain() {
             }
         }
 
+        void cancelParent() {
+            if ((once.get() & ABANDONED) == 0) {
+                parent.cancel(key);
+            }
+        }
+
+        boolean tryAbandon() {
+            return once.get() == FRESH && once.compareAndSet(FRESH, ABANDONED);
+        }
+
         boolean checkTerminated(boolean d, boolean empty, Observer<? super T> a, boolean delayError) {
             if (cancelled.get()) {
                 queue.clear();
-                parent.cancel(key);
                 actual.lazySet(null);
+                cancelParent();
                 return true;
             }
 

src/test/java/io/reactivex/rxjava3/flowable/FlowableGroupByTests.java

@@ -102,7 +102,13 @@ public Flowable<Integer> apply(GroupedFlowable<Integer, Integer> v) {
             }
         }).subscribe(ts);
 
-        ts.assertValues(0, 5, 10, 15, 1, 6, 11, 16, 2, 7, 12, 17, 3, 8, 13, 18, 4, 9, 14, 19);
+        // Behavior change: this now counts as group abandonment because concatMap
+        // doesn't subscribe to the 2nd+ emitted groups immediately
+        ts.assertValues(
+                0, 5, 10, 15, // First group is okay
+                // any other group gets abandoned so we get 16 one-element group
+                1, 2, 3, 4, 6, 7, 8, 9, 11, 12, 13, 14, 16, 17, 18, 19
+                );
         ts.assertComplete();
         ts.assertNoErrors();
     }