Exception not thrown from interval with blank onError

[dvtomas]

Hi all,
this one (and probably some other similar bugs) has been biting me badly last couple of weeks.

RxScala 0.20.0-RC5. Let's have

import rx.lang.scala.Observable
import scala.concurrent.duration._
def fail(o: Observable[Long]) = o.map(x ⇒ if (x > 1) throw new RuntimeException("Bang!"))

then

fail(Observable.items(1, 2, 3)).subscribe(println(_))

throws a RuntimeException as expected, but

fail(Observable.interval(100.milli)).subscribe(println(_))
Thread.sleep(1000)

does not. I'd expect it to throw an exception as well, albeit probably in a different thread.

Thank you, best regards,
Tomáš Dvořák

[benjchristensen]

The exception is getting thrown but FutureTask is swallowing it. It's not obvious to me as to what needs to change to make the ExecutorScheduler throw outside the Callable/FutureTask. Will need to spend more time.

[dvtomas]

OK, thank you. Any progress on this issue will be greatly appreciated.

[benjchristensen]

Related to #1682

[benjchristensen]

@dvtomas I have finished #1682. There isn't much that can be done other than output logs to System.err if exceptions are allowed to propagate without anything catching them (as your examples do) but at least now they won't fail silently.

Can you please review #1682 and let me know if you have any better suggestions?

Here is an example:

        Observable.interval(100, TimeUnit.MILLISECONDS).map(i -> {
            throw new RuntimeException("fail!");
        }).subscribe();

        Thread.sleep(1000);

now outputs

java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.
    at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:50)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    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:745)
Caused by: rx.exceptions.OnErrorNotImplementedException: fail!
    at rx.Observable$36.onError(Observable.java:7387)
    at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:128)
    at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:97)
    at rx.internal.operators.OperatorMap$1.onError(OperatorMap.java:48)
    at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:56)
    at rx.internal.operators.OnSubscribeTimerPeriodically$1.call(OnSubscribeTimerPeriodically.java:51)
    at rx.Scheduler$Worker$1.call(Scheduler.java:118)
    at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:45)
    ... 7 more
Caused by: java.lang.RuntimeException: fail!
    at ErrorsTest.lambda$0(ErrorsTest.java:10)
    at ErrorsTest$$Lambda$1/1406718218.call(Unknown Source)
    at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
    ... 10 more
Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: Long.class
    at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98)
    at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:56)
    ... 10 more

[dvtomas]

I have had a look at #1682. I haven't tested it because it seems that rxjava-scala RCs haven't been released to Maven yet.

My original use case was that I have some observables I know should never fail. If they do, it means that some unhandled exception has been thrown in my code, and it is definitely a bug that should be resolved. In the same spirit as a bug in ordinary code results in an exception being thrown and thus immediate feedback to the developer, I have expected an exception in reactive code to do the same.

Now that I see all the troubles with this issue, I think I'll change my approach to subscribing to observables. I'll just implement a special error handler for handling these "never-failing" observables, and pass it to them during subscribing. This handler could log the error, or try to somehow pass it to the main thread where it could be raised, or anything else. So, for me, your improvement with stack trace print will be a huge help especially during the transition to the handle-everything approach, but other than that I'll just work around this issue with this special handler.

So, as far as I am concerned, I'm happy with your fix and current state of affairs.

Thank you, regards, @dvtomas

[benjchristensen]

Thanks for the feedback.

[jondwillis]

@benjchristensen I have numerous occurrences of this crash throughout my application. However, I am finding it extremely difficult to hunt down the source of these crashes, since the stack trace does not lead back to my client code. Do you have any tips for tracking the source down, given that I have a pretty large codebase that is using schedulers, intervals, observeOns, etc., all over the place?

[benjchristensen]

I responded to your question in the issue you opened: #2293 (comment)