Crash in RxJava due to an observable error even though error is handled

Question

I'm trying to use RxJava (version 1.1.0) in my android app. I've implemented a simple observable/subscriber pattern to invoke API calls over the network, like so (used 'Single' rather than a fuly-blown 'Observable' simply cause there are no multiple events to report for network calls):

// Observable
Single<Data> observable = Single.create(new Single.OnSubscribe<Data>() {
    @Override
    public void call(SingleSubscriber<? super Data> singleSubscriber) {
        try {
            Response<Data> response = mApi.get();

            singleSubscriber.onSuccess(response.body());
        } catch (Exception e) {
            singleSubscriber.onError(e);
        }
    }
}).subscribeOn(Schedulers.io());

// Subscription
observable
    .doOnSuccess(new Action1<Data>() {
        @Override
        public void call(Data data) {
            Log.d("DBG", "success!");
        }
    })
    .doOnError(new Action1<Throwable>() {
        @Override
        public void call(Throwable throwable) {
            Log.d("DBG", "error!");
        }
    })
    .subscribe();

The normal flow works well - I see the 'success' log and data is received. However, when errors occur (e.g. wifi is down), the app crashes even though the error is handled (the 'error!' log gets reported) with the following exception (trimmed down):

12-16 07:33:41.572 6705-6952/com.d4vidi.app E/DBG: error!
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime: FATAL EXCEPTION: RxComputationThreadPool-4
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime: Process: com.d4vidi.app, PID: 6705
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime: java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:60)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.FutureTask.run(FutureTask.java:237)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.lang.Thread.run(Thread.java:818)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:  Caused by: rx.exceptions.OnErrorNotImplementedException: for
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.Single$7.onError(Single.java:1380)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:159)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:71)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:71)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.internal.operators.OperatorSubscribeOn$1$1$1.onError(OperatorSubscribeOn.java:71)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.Single$1$1.onError(Single.java:105)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at com.d4vidi.app.additem.control.Controller$1.call(Controller.java:27)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at com.d4vidi.app.additem.control.Controller$1.call(Controller.java:18)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.Single$1.call(Single.java:110)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.Single$1.call(Single.java:90)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.Observable.unsafeSubscribe(Observable.java:8098)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422) 
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.FutureTask.run(FutureTask.java:237) 
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152) 
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265) 
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112) 
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587) 
12-16 07:33:41.572 6705-6952/com.d4vidi.app E/AndroidRuntime:     at java.lang.Thread.run(Thread.java:818) 

Isn't this the most fundamental usage of RxJava? Why should it crash, then, even though the error is handled in a doOnError() callback - am I missing something here?

Answer 1

The doOnError doesn't handle exceptions, it allows you to peek into the exception but doesn't affect them in any other way. You still have to observe the error and use one of the onErrorXXX or retry() methods to do something about it, i.e., replace it with a valid value, retry it a few times, etc.

Answer 2

Fiddling around with the API I've managed to find a solution. For some reason, changing the subscriber registration as follows gets the error handler to work as expected:

// Subscription
observable.subscribe(
    new Action1<Data>() {
        @Override
        public void call(Data data) {
            Log.d("DBG", "success!");
        }
    },
    new Action1<Throwable>() {
        @Override
        public void call(Throwable throwable) {
            Log.d("DBG", "error!");
        }
    });

(i.e. I've used subscribe(onSuccess, onError) rather than registering callbacks and calling subscribe() separately).

This is very weird. I'll try to dig in further and perhaps report a bug if necessary.