onErrorFlatMap + OnErrorThrowable

[benjchristensen]

I am working on solving a production error handling use case that needs the ability to handle errors on an Observable acting like a message-bus. This requires it to ignore errors if they occur.

With @headinthebox the idea of onErrorFlatMap evolved, but we now have a decision to make about the implementation.

It's pretty easy to solve the first half and allow onErrorFlatMap to return Observables that are injected into the output Observable but don't onComplete and thus allow the stream to continue.

However, we're looking at whether we can also capture the T value that caused the exception to be thrown so we can achieve use cases like this:

    @Test
    public void spliceAndContinueEmitting() {
        TestSubscriber<String> ts = new TestSubscriber<String>();
        Observable.from(1, 2, 3, 4, 5, 6).map(new Func1<Integer, String>() {

            @Override
            public String call(Integer v) {
                if (v < 2 || v > 5) {
                    return "Value=" + v;
                }
                throw new RuntimeException("error in map function: " + v);
            }

        }).onErrorFlatMap(new Func1<OnErrorThrowable, Observable<String>>() {

            @Override
            public Observable<String> call(OnErrorThrowable t) {
                return Observable.from("Error=" + t.getValue());
            }

        }).subscribe(ts);

        ts.assertTerminalEvent();
        System.out.println(ts.getOnErrorEvents());
        assertEquals(0, ts.getOnErrorEvents().size());
        System.out.println(ts.getOnNextEvents());
        ts.assertReceivedOnNext(Arrays.asList("Value=1", "Error=2", "Error=3", "Error=4", "Error=5", "Value=6"));
    }

This outputs:

[Value=1, Error=2, Error=3, Error=4, Error=5, Value=6]

To accomplish this however we must capture the value and wrap the Throwable in all operators that execute user-provided functions. This results in code like this:

            @Override
            public void onNext(T t) {
                try {
                    o.onNext(transformer.call(t));
                } catch (Throwable e) {
                    onError(new OnErrorThrowable(e, t));
                }
            }

instead of this:

            @Override
            public void onNext(T t) {
                try {
                    o.onNext(transformer.call(t));
                } catch (Throwable e) {
                    onError(e);
                }
            }

The drawback to this is that onError will now receive an OnErrorThrowable if the error comes from a user-provided function. The benefit is that the value associated with the failure is now accessible for debugging, reporting, feedback loops etc.

Operators affected by this are cast, doOnEach, filter, groupBy, map, scan, zip (and surely others) as these all take user-provided functions.

Are there reasons we should not wrap these errors inside an OnErrorThrowable?

[cloudbees-pull-request-builder]

RxJava-pull-requests #823 FAILURE
Looks like there's a problem with this pull request

[benjchristensen]

RxJava-pull-requests #823 FAILURE

These errors are known. Waiting on decision before proceeding.

[daveray]

Seems like a fairly major breaking change, at least for code that handles errors. I could see the extra info being useful, though, at least for "bug" exceptions you're not expecting. Intentional exceptions should hopefully already be conveying this info.

Anyway, seems reasonable as long as this behavior's well documented in every affected operator. Blocking operations that turn onError back into throw should also document that calling code may need to do something special to handle exceptions they may be expecting. What's another layer of exception unwrapping? :)

[jstnbckr]

This change will help address the exact use case mentioned above, when an unexpected/untrapped exception occurs in an observable sequence. Rather than shutdown the sequence, the change will allow the observable to remain 'open'. This is especially useful for 'system-oriented' use cases where the observable should not close even if an unexpected/untrapped error occurs. For example, an observable sequence processing user requests for dispatching, or an observable sequence accepting job requests for scheduling. Just because a request within the observable sequence may fail, for some unexpected reason, the 'system-oriented' sequence should remain available for future requests. Also, for debugging and troubleshooting, having the element/item that caused the exception is critical.

[benjchristensen]

I have committed a change that allows this to work without wrapping. Instead if adds a final cause to the Throwable being emitted via onError and then if onErrorFlatMap is used it can derive the value from the causal chain.

A stack trace would end up with something like this at the end of it:

Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: fail

Any problems with this approach? Is this preferred to wrapping?

[cloudbees-pull-request-builder]

RxJava-pull-requests #831 SUCCESS
This pull request looks good

[benjchristensen]

No complains so moving forward with this. Using the causal chain instead of wrapping should be low impact, not intrude on normal uses of Throwables (such as instanceof checks) while still allowing this new behavior and improved stack traces for tracking down what caused the exceptions.

[cloudbees-pull-request-builder]

RxJava-pull-requests #832 SUCCESS
This pull request looks good