2.x: Fix concurrent clear in observeOn while output-fused (#6710)

Author: akarnokd

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

@@ -154,7 +154,7 @@ public final void cancel() {
             upstream.cancel();
             worker.dispose();
 
-            if (getAndIncrement() == 0) {
+            if (!outputFused && getAndIncrement() == 0) {
                 queue.clear();
             }
         }

src/main/java/io/reactivex/internal/operators/observable/ObservableObserveOn.java

@@ -145,7 +145,7 @@ public void dispose() {
                 disposed = true;
                 upstream.dispose();
                 worker.dispose();
-                if (getAndIncrement() == 0) {
+                if (!outputFused && getAndIncrement() == 0) {
                     queue.clear();
                 }
             }

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

@@ -14,6 +14,7 @@
 package io.reactivex.internal.operators.flowable;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.*;
@@ -34,6 +35,7 @@
 import io.reactivex.internal.operators.flowable.FlowableObserveOn.BaseObserveOnSubscriber;
 import io.reactivex.internal.schedulers.ImmediateThinScheduler;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
+import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.*;
 import io.reactivex.schedulers.*;
@@ -1940,4 +1942,37 @@ public void workerNotDisposedPrematurelyNormalInAsyncOutConditional() {
 
         assertEquals(1, s.disposedCount.get());
     }
+
+    @Test
+    public void fusedNoConcurrentCleanDueToCancel() {
+        for (int j = 0; j < TestHelper.RACE_LONG_LOOPS; j++) {
+            List<Throwable> errors = TestHelper.trackPluginErrors();
+            try {
+                final UnicastProcessor<Integer> up = UnicastProcessor.create();
+
+                TestObserver<Integer> to = up.hide()
+                .observeOn(Schedulers.io())
+                .observeOn(Schedulers.single())
+                .unsubscribeOn(Schedulers.computation())
+                .firstOrError()
+                .test();
+
+                for (int i = 0; up.hasSubscribers() && i < 10000; i++) {
+                    up.onNext(i);
+                }
+
+                to
+                .awaitDone(5, TimeUnit.SECONDS)
+                ;
+
+                if (!errors.isEmpty()) {
+                    throw new CompositeException(errors);
+                }
+
+                to.assertResult(0);
+            } finally {
+                RxJavaPlugins.reset();
+            }
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableObserveOnTest.java

@@ -14,6 +14,7 @@
 package io.reactivex.internal.operators.observable;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.*;
@@ -28,7 +29,7 @@
 import io.reactivex.Observer;
 import io.reactivex.annotations.Nullable;
 import io.reactivex.disposables.*;
-import io.reactivex.exceptions.TestException;
+import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.operators.flowable.FlowableObserveOnTest.DisposeTrackingScheduler;
@@ -813,4 +814,36 @@ public void workerNotDisposedPrematurelyNormalInAsyncOut() {
         assertEquals(1, s.disposedCount.get());
     }
 
+    @Test
+    public void fusedNoConcurrentCleanDueToCancel() {
+        for (int j = 0; j < TestHelper.RACE_LONG_LOOPS; j++) {
+            List<Throwable> errors = TestHelper.trackPluginErrors();
+            try {
+                final UnicastSubject<Integer> us = UnicastSubject.create();
+
+                TestObserver<Integer> to = us.hide()
+                .observeOn(Schedulers.io())
+                .observeOn(Schedulers.single())
+                .unsubscribeOn(Schedulers.computation())
+                .firstOrError()
+                .test();
+
+                for (int i = 0; us.hasObservers() && i < 10000; i++) {
+                    us.onNext(i);
+                }
+
+                to
+                .awaitDone(5, TimeUnit.SECONDS)
+                ;
+
+                if (!errors.isEmpty()) {
+                    throw new CompositeException(errors);
+                }
+
+                to.assertResult(0);
+            } finally {
+                RxJavaPlugins.reset();
+            }
+        }
+    }
 }