title: Error handling when using Observable.flatMap
body: <p>There is a sample program below that replicates my issue.</p>
  
  <p>Issue:</p>
  
  <ol>
  <li>Apply the <code>flatMap</code> transformation to some <code>Observable</code></li>
  <li>Subscribe to the aforementioned <code>Observable</code>, store the subscription somewhere</li>
  <li>Dispose of the aforementioned subscription before the <code>Observable</code> terminates naturally</li>
  <li>In an <code>Observable</code> returned by the mapper function, raise an <code>Exception</code></li>
  <li>Flatmap operator doesn't know how to handle the <code>Exception</code>, raises it, program crashes/quits</li>
  </ol>
  
  <p>Preferred/expected behaviour:</p>
  
  <ul>
  <li>The error should be propagated to my <code>onError</code> handler instead of crashing the program when <code>RxJavaPlugins#onError</code> is invoked</li>
  </ul>
  
  <p>The culprit is the snippet of code below, found in <code>ObservableFlatMap</code>. The issue is that once the parent has been disposed of, invocations to <code>addThrowable</code> return false. Thus, the error is never propagated down to <code>onError</code>.</p>
  
  <pre><code>@Override
  public void onError(Throwable t) {
    if (parent.errors.addThrowable(t)) {
        if (!parent.delayErrors) {
            parent.disposeAll();
        }
        done = true;
        parent.drain();
    } else {
        RxJavaPlugins.onError(t);
    }
  }
  </code></pre>
  
  <p>What can I do in this situation? I need an operator that acts like <code>flatMap</code> and propagates errors down to my <code>onError</code> handler instead of crashing my program.</p>
  
  <p>This is a real scenario with an Android app of mine. Subscriptions are automatically disposed when the user exits a window/activity and exceptions may be raised after disposal due to <code>InterruptedIOException</code>s.</p>
  
  <p>Code to replicate issue</p>
  
  <pre><code>import io.reactivex.Observable;
  import io.reactivex.disposables.Disposable;
  import io.reactivex.plugins.RxJavaPlugins;
  import io.reactivex.schedulers.Schedulers;
  
  public class Main {
  
    public static void main(String[] args) throws InterruptedException {
      RxJavaPlugins.setErrorHandler((throwable)-&gt;{
        System.out.println("Please don't come through here");
        throwable.printStackTrace();
      });
      Disposable disposable = Observable.just(1)
          .subscribeOn(Schedulers.computation())
          .flatMap((item)-&gt;{
            return Observable.just(1)
                .doOnNext((arg)-&gt;Thread.sleep(1000))
                .doOnNext((arg)-&gt;{
                  throw new RuntimeException("Error");
                });
          })
          .subscribe(System.out::println, (throwable)-&gt;{
            System.out.println("Please come through here");
            throwable.printStackTrace();
          });
      Thread.sleep(500);
      disposable.dispose();
      Thread.sleep(1000);
    }
  
  }
  </code></pre>
  
  <p>Execution output</p>
  
  <pre><code>Please don't come through here
  io.reactivex.exceptions.UndeliverableException: java.lang.InterruptedException: sleep interrupted
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:349)
    at io.reactivex.internal.operators.observable.ObservableFlatMap$InnerObserver.onError(ObservableFlatMap.java:573)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:119)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:119)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onNext(ObservableDoOnEach.java:99)
    at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarDisposable.run(ObservableScalarXMap.java:248)
    at io.reactivex.internal.operators.observable.ObservableJust.subscribeActual(ObservableJust.java:35)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.subscribeInner(ObservableFlatMap.java:162)
    at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.onNext(ObservableFlatMap.java:139)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onNext(ObservableSubscribeOn.java:58)
    at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarDisposable.run(ObservableScalarXMap.java:248)
    at io.reactivex.internal.operators.observable.ObservableJust.subscribeActual(ObservableJust.java:35)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
    at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38)
    at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:748)
  Caused by: java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at Main.lambda$null$1(Main.java:18)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onNext(ObservableDoOnEach.java:95)
    ... 22 more
  </code></pre>
  
  <p>Expected/preferred output</p>
  
  <pre><code>Please come through here
  java.lang.RuntimeException: Error
    at Main.lambda$null$2(Main.java:19)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onNext(ObservableDoOnEach.java:95)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onNext(ObservableDoOnEach.java:103)
    at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarDisposable.run(ObservableScalarXMap.java:248)
    at io.reactivex.internal.operators.observable.ObservableJust.subscribeActual(ObservableJust.java:35)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.subscribeInner(ObservableFlatMap.java:162)
    at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.onNext(ObservableFlatMap.java:139)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onNext(ObservableSubscribeOn.java:58)
    at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarDisposable.run(ObservableScalarXMap.java:248)
    at io.reactivex.internal.operators.observable.ObservableJust.subscribeActual(ObservableJust.java:35)
    at io.reactivex.Observable.subscribe(Observable.java:10903)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
    at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38)
    at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:748)
  </code></pre>