3.x: Fix concatMapDelayError not continuing on fused inner source crash (#6523)

Author: akarnokd

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

@@ -519,10 +519,13 @@ void drain() {
                                     vr = supplier.get();
                                 } catch (Throwable e) {
                                     Exceptions.throwIfFatal(e);
-                                    upstream.cancel();
                                     errors.addThrowable(e);
-                                    downstream.onError(errors.terminate());
-                                    return;
+                                    if (!veryEnd) {
+                                        upstream.cancel();
+                                        downstream.onError(errors.terminate());
+                                        return;
+                                    }
+                                    vr = null;
                                 }
 
                                 if (vr == null) {

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

@@ -15,14 +15,14 @@
 
 import static org.junit.Assert.assertEquals;
 
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 
 import io.reactivex.*;
-import io.reactivex.exceptions.TestException;
+import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.operators.flowable.FlowableConcatMap.WeakScalarSubscription;
 import io.reactivex.schedulers.Schedulers;
@@ -169,4 +169,42 @@ public void run() throws Exception {
 
         assertEquals(0, counter.get());
     }
+
+    @Test
+    public void delayErrorCallableTillTheEnd() {
+        Flowable.just(1, 2, 3, 101, 102, 23, 890, 120, 32)
+        .concatMapDelayError(new Function<Integer, Flowable<Integer>>() {
+          @Override public Flowable<Integer> apply(final Integer integer) throws Exception {
+            return Flowable.fromCallable(new Callable<Integer>() {
+              @Override public Integer call() throws Exception {
+                if (integer >= 100) {
+                  throw new NullPointerException("test null exp");
+                }
+                return integer;
+              }
+            });
+          }
+        })
+        .test()
+        .assertFailure(CompositeException.class, 1, 2, 3, 23, 32);
+    }
+
+    @Test
+    public void delayErrorCallableEager() {
+        Flowable.just(1, 2, 3, 101, 102, 23, 890, 120, 32)
+        .concatMapDelayError(new Function<Integer, Flowable<Integer>>() {
+          @Override public Flowable<Integer> apply(final Integer integer) throws Exception {
+            return Flowable.fromCallable(new Callable<Integer>() {
+              @Override public Integer call() throws Exception {
+                if (integer >= 100) {
+                  throw new NullPointerException("test null exp");
+                }
+                return integer;
+              }
+            });
+          }
+        }, 2, false)
+        .test()
+        .assertFailure(NullPointerException.class, 1, 2, 3);
+    }
 }