Rxjava: 2.x: Uncaught errors fail silently in junit tests.

[ZacSweers]

I was writing a utility that involved decorating observers and catching uncaught errors from the delegate observer's onError handling when I noticed in testing that tests always passed. There would be log output, but tests would pass none the less.

Demo of the issue can be found here, but the gist of it is that the following test passes in standard JUnit:

@Test
public void blerg() {
    Observable.error(new RuntimeException("This should fail the test"))
            .subscribe();
}

I'm trying to understand why errors are passed directly to the handler rather than just thrown, and also curious to get thoughts on making this behavior configurable (if for no other reason than to disable in junit tests). I can't find any details about why this behavior happens in JUnit either.

I think this has worrisome implications. In our codebase alone, I dropped in an error watching rule to track these and discovered over 150 tests with uncaught exceptions that are being marked as passing otherwise.

[0kai]

you may use TestObserver to subscribe the result

[akarnokd]

Unlike 1.x, OnErrorNotImplementedException is not thrown. Also please read on the error handling section of the wiki.

[ZacSweers]

I did read that, but it doesn't really explain why it deals directly with the current thread's UncaughtExceptionHandler rather than just throwing if there's no onError hook set.

@0kai I'm not trying to figure out how to test errors, rather I'm trying to make sure that uncaught exceptions in tests aren't marked as passing. At least not silently anyway

[akarnokd]

2.x Subscribers and Observers must not throw from their onXXX methods. Install an error handler before the test and verify there was no undeliverable error. See how we do it in RxJava tests:

https://github.com/ReactiveX/RxJava/blob/3bfc275ca2f8d14a67305bbadb7bbc6315205250/src/test/java/io/reactivex/parallel/ParallelPeekTest.java#L55

[ZacSweers]

I think I'm not being clear. I'm not trying to test if onXXX methods throw. I'm concerned that errors in the stream with no error handling don't actually fail tests because they're sent directly to the handler rather than just throwing if there's no plugin error handler. Trying to understand why we send directly to the handler.

[akarnokd]

Because you can't throw in a push/async environment and expect it to try-catch it.

[arturdryomov]

We have a very similar issue.

The Thread.UncaughtExceptionHandler documentation states the following.

When a thread is about to terminate due to an uncaught exception the Java Virtual Machine will query the thread for its UncaughtExceptionHandler using getUncaughtExceptionHandler() and will invoke the handler's uncaughtException method, passing the thread and the exception as arguments.

If I understand this correctly, the direct throw will eventually pass the exception to UncaughtExceptionHandler anyway. It seems like the direct pass to UncaughtExceptionHandler without throw will not terminate the current thread. @akarnokd, was this a motivation for not rethrowing the error and using the direct pass to UncaughtExceptionHandler?

Most likely this is not an issue for real-world environments, but this affects unit tests we run on JVM using JUnit.

import org.junit.Test
import io.reactivex.Observable as RxJava2Observable
import rx.Observable as RxJava1Observable

class RxError {

    @Test fun `RxJava 1 test failure with OnErrorNotImplementedException as expected`() {
        RxJava1Observable.fromCallable { throw RuntimeException() }.subscribe()
    }

    @Test fun `RxJava 2 test success, just prints OnErrorNotImplementedException stacktrace not as expected`() {
        RxJava2Observable.fromCallable { throw RuntimeException() }.subscribe()
    }

}

CC @artem-zinnatullin

[akarnokd]

RxJava 2 only passes fatal exceptions upwards, which end up in the uncaught exception handler. Any other exception goes through onError or RxJavaPlugins.onError. RxJavaPlugins.onError() prints the stacktrace and passes the error to the uncaught exception handler of the current thread. This is due to the Reactive-Streams specification which forbids throwing anything non-fatal from Subscribers and by mirror from Observers as well.

RxJava 2 test success, just prints OnErrorNotImplementedException stacktrace not as expected

What stacktrace did you expect?

[arturdryomov]

What stacktrace did you expect?

Sorry for the poor wording. I meant that the expected behaviour from my side is to fail a test due to uncaught exception

[akarnokd]

This is why RxJava's own test often utilize TestHelper.trackPluginErrors() and after the run we check for any errors in the List<Throwable> returned by that method.

[artem-zinnatullin]

Yeah, this is very serious issue for us as well (sad that we started migration that late…)

@akarnokd theoretically what if RxJavaPlugins.onError() would actually throw passed OnErrorNotImplementedException?

For RS subscriber everything will be ok, since OnErrorNotImplementedException is RxJava v2 specific type and StrictSubscriber does not throw it (only if user would do this manually), that would allow RxJava v2 keep compatibility with Reactive Streams and pass TCK.

Basically similar to how it works now with rule 3.9 #5274 and few other cases when if you use RxJava v2 specific types it can break specification, but if you use RS types — it follows RS.

This is kinda breaking change, but not really, since in production ~no one wraps rx chains with try-catch "Because you can't throw in a push/async environment and expect it to try-catch it." and exceptions without try-catch will achieve thread uncaught handler normally as before while in tests users will actually see problems since test frameworks wrap tests with try-catch and reactive chains in tests are mostly synchronous.

Plus we could attach RxJavaPlugin sample that imitates previous behavior to the changelog.

[akarnokd]

The RxJavaPlugins allows you to specify a global handler which you can override temporarily with a @Rule and check for excess errors. The current logic doesn't throw fatal exceptions upwards either so OnErrorNotImplemented wouldn't travel upwards.

[artem-zinnatullin]

Right, but users will have to put @Rule in each test class, not everybody use JUnit (we also use two versions of Spek in our projects and mix JUnit4 with JUnit5), I haven't found a way to set global custom JUnit4 test runner in Gradle and by default IntelliJ runs tests on its own using build system only to generate test class files so it'll also require some customization.

"In production" you typically have an entrypoint (Application classes exist in most of both in front- and backend frameworks) which is a convenient place to hook into RxJavaPlugins and set desired behavior.

While tests are usually isolated from each other and each class if not method will require customization for um, detecting errors… which is basically what each test should do by default… It just seems wrong to not inverse current behavior.

@akarnokd what are your concerns about throwing OnErrorNotImplementedException for RxJava v2 specific subscribers and not doing so for Reactive Streams subscribers?

Mine is that it can create sort of unexpected behavior if you'll try to mix RxJava v2 with some other RS implementations in compare to how RxJava v2 will work in isolation, but I'm not sure that mixing is common use case. I'm much more afraid that current behavior is very unexpected for v1 users who are migrating to v2.

[akarnokd]

Throwing anything but fatal exceptions is undefined behavior.

I guess you either manually write those unit tests or have some sort of generator. If written manually, you can always compose with extra checks before subscribing to a source. If generated, I assume the generator is written manually since I don't think any testing framework has first class support for RxJava yet.

Add the plugin error tracking via doOnSubscribe() and remove it via doFinally.

public static <T> TestObserver<T> createTester(Observable<T> source, Observer<T> mocked) {

    TestObserver<T> to = mocked != null ? new TestObserver<>(mocked) : new TestObserver<>();

    List<Throwable> list = Collections.synchronizedList(new ArrayList<>());

    return source.doOnSubscribe(s -> {
        RxJavaPlugins.reset();
        RxJavaPlugins.setErrorHandler(e -> list.add(e));
    })
    .doFinally(() -> {
        RxJavaPlugins.reset();
        for (Throwable e : list) {
           to.onError(e);
        }
    })
    .subscribeWith(to);
}

[artem-zinnatullin]

Throwing anything but fatal exceptions is undefined behavior.

In RxJava v1 OnErrorNotImplementedException was considered as fatal, what makes it non-fatal for RxJava v2?

[akarnokd]

It wasn't in the original set of fatal exceptions of 2.x and throwing from Reactive-Streams is somewhat of a gamble. For example, a multi-library flow may not consider it as fatal or swallow it completely anyway (but most will consider an OutOfMemoryError as such btw). The most reliable way for exceptions is down where RxJavaPlugins.onError is at the very bottom.

I believe RxJava already offers the necessary hooking capabilities to detect such errors in tests; most likely you have to mock out schedulers already, making the test synchronous and at the end, you can synchronously check any excess errors with an approach like I showed above.

[artem-zinnatullin]

throwing from Reactive-Streams is somewhat of a gamble

That's what I want to avoid and throw only if RxJava v2 specific subscriber was used.

OnErrorNotImplementedException can not be thrown in the middle of rx chain. Only by calling v2-specificsubscribe() without error handler, which reduces possibility of multi-library problems since they will use RS org.reactivestreams.Publisher.subscribe(org.reactivestreams.Subscriber).

As said before, v2 violates some RS rules if v2 specific subscriber was used, but respects RS if RS subscriber comes. And I can create problematic multi-library flow because of that, but only if I'd like to.

---

For example, a multi-library flow may not consider it as fatal or swallow it completely anyway

Btw, this is already happening with RxJava v1 and v2 connected through interop library, v2 swallows OnErrorNotImplemented thrown by v1:

@Test
fun interopProblem() {
    try {
        Observable2
                .fromCallable { throw RuntimeException("Test") }
                .toObservable1()
                .subscribe()
        fail("Should throw rx.exceptions.OnErrorNotImplementedException")
    } catch (expected: rx.exceptions.OnErrorNotImplementedException) {
        // ok.
    }
}

---

most likely you have to mock out schedulers already, making the test synchronous and at the end, you can synchronously check any excess errors with an approach like I showed above.

Good point, but.

Scheduling is explicit, our code will not compile if we won't pass scheduler (either real or test one).

While current design of error handling is implicit and not only allows tests to compile without checks suggested above, but also returns normally and tests pass as false positives.

[akarnokd]

.subscribe()

There is a lint check for such use cases: https://bitbucket.org/littlerobots/rxlint

[artem-zinnatullin]

We normally don't have errors in streams and convert them to values using Kotlin sealed classes, so we don't use subscribe() overloads with error handling, rxlint won't help (and it's Android specific).

But we need to detect unexpected errors in tests.

I believe RxJava already offers the necessary hooking capabilities to detect such errors in tests

And we have 3k+ test cases written with different test frameworks, hooking into each test class is manual and error-prone work. There are hundreds if not thousands other projects migrating from v1 to v2 that'll have to do the same and this number will only grow.

[akarnokd]

Write a custom Subscriber that throws from its onError method and subscribe with that instead of the empty subscribe().

[artem-zinnatullin]

Please correct me if I'm wrong but:

1. Throwing OnErrorNotImplementedException is not undefined behavior since this exception can be thrown in known limited set of cases.
2. Throwing OnErrorNotImplementedException does not break compatibility with Reactive Streams since it can be thrown only if RxJava v2 specific subscribe() method was called and not if RS Publisher.subscribe() was called.
3. Throwing OnErrorNotImplementedException is not really a breaking change since if user has no try/catch → it'll be delivered to UncaughtExceptionHandler as before and handling OENIE through custom RxJavaPlugins.onError() is typically wrong design and should be done via correct onError() or subscribe() with error handler on a stream level.

Which makes possible to do this change in 2.x and allow users test Rx code as before with 1.x.

[arturdryomov]

@hzsweers Just curious — how did you solve this issue on your side?

[ZacSweers]

We just have a junit rule that installs an error hook via rxjava plugins that throws it

[ZacSweers]

This is actually a problem outside of tests too unfortunately. Say I'm on Android and I want to write a custom plugin for decorating observers (such as tracking analytics). I want to give the delegate observer a shot at onError and try/catch to react to how it handled it, but if it's the default rxjava execution and hands it over to the current thread's uncaught exception handler via its internal uncaught() method, the app will just quite immediately without the catch clause getting a chance.

Installing a custom error handler via plugins is not an option either, as it is just try/catch'd and pipes the error again through the above uncaught() pipeline.

Could I propose making this configurable via system property? Something like rx2.uncaughtBehavior with values "default" and "rethrow"?

Edit: Proposed a strawman impl in #5569

[akarnokd]

I'd say this use case has now a workaround with #5590 so closing it.

[almozavr]

Would be great to have a snippet for such a workaround if any.

[ZacSweers]

https://github.com/uber/AutoDispose/blob/master/autodispose/src/test/java/com/uber/autodispose/RxErrorsRule.java

[ZacSweers]

^ for tests anyway. For prod, you can basically install an observer subscribe hook and watch for those interfaces

[almozavr]

@hzsweers thanks! Still feels like a bit of complex solution and hardly portable to e.g. Spek :(

[JorgeCastilloPrz]

Thanks for the Rule, @hzsweers. Not super cool to use something like this, but no choice. It's something.

[saket]

Permalink to the current version of RxErrorsRule for others: https://github.com/uber/AutoDispose/blob/6e1399512f132bc5052c4b1ba111debb6eb430e1/test-utils/src/main/java/com/uber/autodispose/test/RxErrorsRule.java

[ZacSweers]

I'd like to re-kick the proposal for a property or hook to configure this. While the workarounds of the errors rule for tests and LambdaConsumerIntrospection for runtime avoidance, there is actually a third case I've hit today that neither of them cover:

Say you have a delegating observer, such as what we have in AutoDispose. In an environment where the uncaught exception handler just calls System.exit() (i.e. Android), it becomes impossible to try-catch delegate on____ calls if there is no error handling. Not only can the original exception not be caught, but neither can the OnErrorNotImplementedException, and there's nothing that can safely be done about it from the consumer standpoint.

In #5569 part of the discussion proposals was something like a RxJavaPlugins.setOnErrorNotImplementedConsumer(Function<Consumer<Throwable>, Consumer<Throwable>>). Would you be open to that? Specifically, I think it's really important to be able to re-route something to ensure it doesn't hit the current uncaught path and thus becoming un-try-catchable

[akarnokd]

Okay, let's see the proposed code changes.

[ZacSweers]

PR opened: https://github.com/ReactiveX/RxJava/pull/6378

[rjrjr]

https://github.com/vRallev just solved this for us in a pretty clean way at Square: introduced a simple jvmAgent that sets up a default uncaught exception handler, and some gradle tweaking to make sure that our unit tests all depend on the module that puts the agent into effect.

I plan to steal the technique for https://github.com/square/workflow, will try to remember to ping here when I do so.

[jameswald]

We're using that java agent technique but JVM tests still seem to pass in some cases. At least we get this in test output:

Exception: com.example.test.agent.UncaughtException thrown from the UncaughtExceptionHandler in thread "ModernAsyncTask #3"