3.x: Fix Flowable.flatMap not canceling the inner sources on outer error (#6826)

Author: akarnokd

Diff for: src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMap.java

@@ -321,6 +321,11 @@ public void onError(Throwable t) {
             }
             if (errors.tryAddThrowableOrReport(t)) {
                 done = true;
+                if (!delayErrors) {
+                    for (InnerSubscriber<?, ?> a : subscribers.getAndSet(CANCELLED)) {
+                        a.dispose();
+                    }
+                }
                 drain();
             }
         }

Diff for: src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMapTest.java

@@ -1116,4 +1116,38 @@ public Publisher<Integer> apply(Integer v) throws Throwable {
             }
         });
     }
+
+    @Test
+    public void mainErrorsInnerCancelled() {
+        PublishProcessor<Integer> pp1 = PublishProcessor.create();
+        PublishProcessor<Integer> pp2 = PublishProcessor.create();
+
+        pp1
+        .flatMap(v -> pp2)
+        .test();
+
+        pp1.onNext(1);
+        assertTrue("No subscribers?", pp2.hasSubscribers());
+
+        pp1.onError(new TestException());
+
+        assertFalse("Has subscribers?", pp2.hasSubscribers());
+    }
+
+    @Test
+    public void innerErrorsMainCancelled() {
+        PublishProcessor<Integer> pp1 = PublishProcessor.create();
+        PublishProcessor<Integer> pp2 = PublishProcessor.create();
+
+        pp1
+        .flatMap(v -> pp2)
+        .test();
+
+        pp1.onNext(1);
+        assertTrue("No subscribers?", pp2.hasSubscribers());
+
+        pp2.onError(new TestException());
+
+        assertFalse("Has subscribers?", pp1.hasSubscribers());
+    }
 }

Diff for: src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableFlatMapTest.java

@@ -26,7 +26,7 @@
 import io.reactivex.rxjava3.core.*;
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.core.Observer;
-import io.reactivex.rxjava3.disposables.*;
+import io.reactivex.rxjava3.disposables.Disposable;
 import io.reactivex.rxjava3.exceptions.*;
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.functions.Functions;
@@ -1079,4 +1079,38 @@ public Observable<Integer> apply(Integer v) throws Throwable {
             }
         });
     }
+
+    @Test
+    public void mainErrorsInnerCancelled() {
+        PublishSubject<Integer> ps1 = PublishSubject.create();
+        PublishSubject<Integer> ps2 = PublishSubject.create();
+
+        ps1
+        .flatMap(v -> ps2)
+        .test();
+
+        ps1.onNext(1);
+        assertTrue("No subscribers?", ps2.hasObservers());
+
+        ps1.onError(new TestException());
+
+        assertFalse("Has subscribers?", ps2.hasObservers());
+    }
+
+    @Test
+    public void innerErrorsMainCancelled() {
+        PublishSubject<Integer> ps1 = PublishSubject.create();
+        PublishSubject<Integer> ps2 = PublishSubject.create();
+
+        ps1
+        .flatMap(v -> ps2)
+        .test();
+
+        ps1.onNext(1);
+        assertTrue("No subscribers?", ps2.hasObservers());
+
+        ps2.onError(new TestException());
+
+        assertFalse("Has subscribers?", ps1.hasObservers());
+    }
 }