3.x: Fix scheduled tasks' fatal exception behavior

src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTask.java

@@ -16,7 +16,6 @@
 
 package io.reactivex.rxjava3.internal.schedulers;
 
-import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 
 /**
@@ -39,7 +38,7 @@ public void run() {
             runnable.run();
             runner = null;
         } catch (Throwable ex) {
-            Exceptions.throwIfFatal(ex);
+            // Exceptions.throwIfFatal(ex); nowhere to go
             runner = null;
             lazySet(FINISHED);
             RxJavaPlugins.onError(ex);

src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectTask.java

@@ -18,6 +18,8 @@
 
 import java.util.concurrent.Callable;
 
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
+
 /**
  * A Callable to be submitted to an ExecutorService that runs a Runnable
  * action and manages completion/cancellation.
@@ -35,10 +37,15 @@ public ScheduledDirectTask(Runnable runnable) {
     public Void call() {
         runner = Thread.currentThread();
         try {
-            runnable.run();
-        } finally {
-            lazySet(FINISHED);
-            runner = null;
+            try {
+                runnable.run();
+            } finally {
+                lazySet(FINISHED);
+                runner = null;
+            }
+        } catch (Throwable ex) {
+            // Exceptions.throwIfFatal(e); nowhere to go
+            RxJavaPlugins.onError(ex);
         }
         return null;
     }

src/test/java/io/reactivex/rxjava3/schedulers/ComputationSchedulerTests.java

@@ -18,13 +18,14 @@
 import java.util.HashMap;
 import java.util.concurrent.*;
 
-import io.reactivex.rxjava3.disposables.Disposable;
 import org.junit.Test;
 
 import io.reactivex.rxjava3.core.*;
 import io.reactivex.rxjava3.core.Scheduler.Worker;
+import io.reactivex.rxjava3.disposables.Disposable;
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.schedulers.ComputationScheduler;
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 
 public class ComputationSchedulerTests extends AbstractSchedulerConcurrencyTests {
 
@@ -192,4 +193,95 @@ public void run() {
 
         assertEquals(0, calls[0]);
     }
+
+    @Test
+    public void exceptionFromObservableShouldNotBeSwallowed() throws Exception {
+        // Exceptions, fatal or not, should be handled by
+        // #1 observer's onError(), or
+        // #2 RxJava exception handler, or
+        // #3 thread's uncaught exception handler,
+        // and should not be swallowed.
+        try {
+            CountDownLatch latch = new CountDownLatch(1);
+
+            // #3 thread's uncaught exception handler
+            Scheduler computationScheduler = new ComputationScheduler(new ThreadFactory() {
+                @Override
+                public Thread newThread(Runnable r) {
+                    Thread t = new Thread(r);
+                    t.setUncaughtExceptionHandler((thread, throwable) -> {
+                        latch.countDown();
+                    });
+                    return t;
+                }
+            });
+
+            // #2 RxJava exception handler
+            RxJavaPlugins.setErrorHandler(h -> {
+                latch.countDown();
+            });
+
+            // #1 observer's onError()
+            Observable.create(s -> {
+
+                s.onNext(1);
+
+                if (true) {
+                    throw new OutOfMemoryError();
+                }
+
+                s.onComplete();
+
+            }).subscribeOn(computationScheduler)
+            .subscribe(v -> {
+            }, e -> {
+                latch.countDown();
+            });
+
+            assertTrue(latch.await(2, TimeUnit.SECONDS));
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void exceptionFromObserverShouldNotBeSwallowed() throws Exception {
+        // Exceptions, fatal or not, should be handled by
+        // #1 observer's onError(), or
+        // #2 RxJava exception handler, or
+        // #3 thread's uncaught exception handler,
+        // and should not be swallowed.
+        try {
+            CountDownLatch latch = new CountDownLatch(1);
+
+            // #3 thread's uncaught exception handler
+            Scheduler computationScheduler = new ComputationScheduler(new ThreadFactory() {
+                @Override
+                public Thread newThread(Runnable r) {
+                    Thread t = new Thread(r);
+                    t.setUncaughtExceptionHandler((thread, throwable) -> {
+                        latch.countDown();
+                    });
+                    return t;
+                }
+            });
+
+            // #2 RxJava exception handler
+            RxJavaPlugins.setErrorHandler(h -> {
+                latch.countDown();
+            });
+
+            // #1 observer's onError()
+            Flowable.interval(500, TimeUnit.MILLISECONDS, computationScheduler)
+                    .subscribe(v -> {
+                        throw new OutOfMemoryError();
+                    }, e -> {
+                        latch.countDown();
+                    });
+
+            assertTrue(latch.await(2, TimeUnit.SECONDS));
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }