java.lang.Error is not propagated through onError (2.x)

[Auffahrend]

RX Version: io.reactivex.rxjava2:rxjava:2.1.8
JDK Version Oracle 1.8.0_151
Summary: Observable, Single, Completable, etc do not always propagate errors (subclasses of java.lang.Error) and subsequently do not complete, which causes deadlocks and hanging.

Details: My colleagues and I found out that some types of errors within observable (as well as Single, Completable) do not call Emitter::onError. I was able to reproduce it with errors like IllegalAccessError, StackOverflowError, OutOfMemoryError, etc. But some errors do not cause hanging, e.g. IOError.
Originally, the bug was found because Singe.merge() was hanging on non-blocking method throwing java.lang.NoSuchMethodError due to a method reference to a generic type union (which is another story).

Note, that if you .subscribeOn(Schedulers.io()) the error is emitted downstream. Hanging occurs only when you .subscribeOn(Schedulers.computation()) or with a custom limited scheduler, e.g. Schedulers.from(Executors.newFixedThreadPool(50)).

How to reproduce:

1. With Observable:

    public static void main(String[] args) {
        Observable.fromCallable(() -> { throw new OutOfMemoryError("FATAL!"); })
                .subscribeOn(Schedulers.computation())
                .blockingFirst();
        System.out.println("Finished");
    } 

2. With Single:

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        Single.fromCallable(() -> { throw new OutOfMemoryError("FATAL!"); })
                .subscribeOn(Schedulers.computation())
                .subscribe(v -> latch.countDown(), e -> latch.countDown());
        latch.await();
        System.out.println("Finished");
    }

3. With Completable:

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        Completable.fromCallable(() -> { throw new OutOfMemoryError("FATAL!"); })
                .subscribeOn(Schedulers.computation())
                .subscribe(latch::countDown);
        latch.await();
        System.out.println("Finished");
    }

4. With Maybe:

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        Maybe.fromCallable(() -> { throw new OutOfMemoryError("FATAL!"); })
                .subscribeOn(Schedulers.computation())
                .subscribe(v -> latch.countDown(), e -> latch.countDown());
        latch.await();
        System.out.println("Finished");
    }

I will be glad to provide any necessary additional information.

[akarnokd]

This is by design. Most Error subclass is considered fatal and is rethrown immediately. StackOverflowError is a subclass of VirtualMachineError so it is not thrown.

[Auffahrend]

Thanks for a quick reply.
So why isn't is rethrown? Why if you subscribe with io(), the pipeline is being completed, and with computation() it is not? How should I modify the code above to make it reliably rethrow Errors?

[akarnokd]

It should be rethrown which will end up in the Thread's uncaught exception handler. Which exact case did signal an onError for you?

How should I modify the code above to make it reliably rethrow Errors?

Catch Errors and repackage them into RuntimeExceptions.

[Auffahrend]

Which exact case did signal an onError for you?

with direct subclasses of Error, the onError is called, e.g.:

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        Single.fromCallable(() -> { throw new IOError(new RuntimeException("FATAL!")); })
                .subscribeOn(Schedulers.computation())
                .subscribe(v -> {
                            latch.countDown();
                            System.out.println(v);
                        },
                        e -> {
                            latch.countDown();
                            throw new RuntimeException(e);
                        });
        latch.await();
        System.out.println("Finished");
    }

You can also try AssertionError, it also doesn't hang.
But with indirect subclasses of Error (e.g. subclasses of LinkageError or VirtualMachineError) it hangs, even with UncaughtExceptionHandler:

    public static void main(String[] args) throws InterruptedException {
        Thread.UncaughtExceptionHandler handler = (t, e) -> {
            System.err.println("Uncaught exception!");
            System.err.println(e);
        };

        Scheduler scheduler = Schedulers.from(Executors.newFixedThreadPool(10, r -> {
            Thread thread = new Thread(r);
            thread.setUncaughtExceptionHandler(handler);
            return thread;
        }));

        CountDownLatch latch = new CountDownLatch(1);
        Single.fromCallable(() -> { throw new AbstractMethodError("FATAL!"); })
                .subscribeOn(scheduler)
                .subscribe(v -> {
                            latch.countDown();
                            System.out.println(v);
                        },
                        e -> {
                            latch.countDown();
                            throw new RuntimeException(e);
                        });
        latch.await();
        System.out.println("Finished");
    }

What do you mean by

Catch Errors and repackage them into RuntimeExceptions.

Wrapping each method call in try {} catch (Throwable) sounds horrible.

[akarnokd]

These are the fatal error classes that don't get propagated (also their subclasses):

• VirtualMachineError
• ThreadDeath
• LinkageError

Don't throw these programmatically. They quit the flow because they are considered fatal and unrecoverable errors. Trying to propagate them as errors will likely lead to more crashing. See #748 which prompted for the now established behavior.

What do you mean by

try {
   // your unreliable method calls here
} catch (VirtualMachineError ex) {
   throw new RuntimeException(ex);
}

But as stated above, when such exceptions are catched by RxJava, they are considered fatal and unrecoverable from the flow's perspective. It becomes unsafe to call any method on downstream. Imagine you have a NoSuchMethodError and you have an unconditional retry in the flow below that.

[Auffahrend]

I see your point. I agree with you completely on the point "don't throw error programatically" and we did not.
But the problem we faced is the JDK bug which causes NoSuchMethodError in otherwise correct code. Currently we are migrating some services on rx, and it scares me to think that at some point some LinkageError or StackOverflowError will not cause crash of application but rather silently kill all computation threads and hang, without signals in log or NewRelic.

If you are curious, the jdk bug is described in https://stackoverflow.com/questions/27031244/lambdaconversionexception-with-generics-jvm-bug and for Oracle's jdk it's reported to be fixed only in jdk9 https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8141508

I still not completely understand how to make application reliably crash on fatal errors, though, but thanks anyway for your help.

[masc3d]

How should I modify the code above to make it reliably rethrow Errors?

Catch Errors and repackage them into RuntimeExceptions.

not really a great solution to have important exceptions potentially go entirely unnoticed unless all consumers explicitly handle this.

why not delegate to global error handler at least and make swallowing behaviour for those errors default, so at least it could be overridden via RxJavaPlugins.setErrorHandler @akarnokd I would propose to reopen this.

[akarnokd]

No. See the reasoning above.

[masc3d]

you provided retry as an example, it's not reasonable.
retry shouldn't retry on those low level errors but only regular exceptions.

other examples?

[masc3d]

I agree that those errors should terminate the flow unconditionally. but what's the reason for not invoking the global error handler and swallowing them entirely?