In ObservableConcatMapSingle, if dispose is called before innerError, forward errors to RxJavaPlugins.onError instead of swallowing them. Fixes ReactiveX#6587

Author: Erlkoenig90

src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingle.java

@@ -123,7 +123,7 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            if (errors.addThrowable(t)) {
+            if (errors.addThrowable(t) && !cancelled) {
                 if (errorMode == ErrorMode.IMMEDIATE) {
                     inner.dispose();
                 }
@@ -148,6 +148,10 @@ public void dispose() {
             if (getAndIncrement() == 0) {
                 queue.clear();
                 item = null;
+
+                if (errors.get() != null) {
+                    RxJavaPlugins.onError(errors.terminate());
+                }
             }
         }
 
@@ -163,7 +167,7 @@ void innerSuccess(R item) {
         }
 
         void innerError(Throwable ex) {
-            if (errors.addThrowable(ex)) {
+            if (errors.addThrowable(ex) && !cancelled) {
                 if (errorMode != ErrorMode.END) {
                     upstream.dispose();
                 }
@@ -188,25 +192,31 @@ void drain() {
             for (;;) {
 
                 for (;;) {
-                    if (cancelled) {
-                        queue.clear();
-                        item = null;
-                        break;
-                    }
-
+                    boolean isCancelled = cancelled;
                     int s = state;
 
                     if (errors.get() != null) {
-                        if (errorMode == ErrorMode.IMMEDIATE
-                                || (errorMode == ErrorMode.BOUNDARY && s == STATE_INACTIVE)) {
-                            queue.clear();
-                            item = null;
-                            Throwable ex = errors.terminate();
-                            downstream.onError(ex);
+                        if (isCancelled) {
+                            RxJavaPlugins.onError(errors.terminate());
                             return;
+                        } else {
+                            if (errorMode == ErrorMode.IMMEDIATE
+                                    || (errorMode == ErrorMode.BOUNDARY && s == STATE_INACTIVE)) {
+                                queue.clear();
+                                item = null;
+                                Throwable ex = errors.terminate();
+                                downstream.onError(ex);
+                                return;
+                            }
                         }
                     }
 
+                    if (isCancelled) {
+                        queue.clear();
+                        item = null;
+                        break;
+                    }
+
                     if (s == STATE_INACTIVE) {
                         boolean d = done;
                         T v = queue.poll();

src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingleTest.java

@@ -212,6 +212,42 @@ protected void subscribeActual(
         }
     }
 
+    @Test
+    public void innerErrorAfterMainDispose() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final PublishSubject<Integer> ps = PublishSubject.create();
+
+            final AtomicReference<SingleObserver<? super Integer>> obs = new AtomicReference<SingleObserver<? super Integer>>();
+
+            TestObserverEx<Integer> to = ps.concatMapSingle(
+                    new Function<Integer, SingleSource<Integer>>() {
+                        @Override
+                        public SingleSource<Integer> apply(Integer v)
+                                throws Exception {
+                            return new Single<Integer>() {
+                                @Override
+                                protected void subscribeActual(
+                                        SingleObserver<? super Integer> observer) {
+                                    observer.onSubscribe(Disposables.empty());
+                                    obs.set(observer);
+                                }
+                            };
+                        }
+                    }
+            ).to(TestHelper.<Integer>testConsumer());
+
+            ps.onNext(1);
+
+            to.dispose();
+            obs.get().onError(new TestException("inner"));
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class, "inner");
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
     @Test
     public void delayAllErrors() {
         TestObserverEx<Object> to = Observable.range(1, 5)