3.x: Fix scheduled tasks' fatal exception behavior

src/main/java/io/reactivex/rxjava3/core/Scheduler.java

@@ -18,11 +18,9 @@
 
 import io.reactivex.rxjava3.annotations.*;
 import io.reactivex.rxjava3.disposables.Disposable;
-import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.functions.Function;
 import io.reactivex.rxjava3.internal.disposables.*;
 import io.reactivex.rxjava3.internal.schedulers.*;
-import io.reactivex.rxjava3.internal.util.ExceptionHelper;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 import io.reactivex.rxjava3.schedulers.SchedulerRunnableIntrospection;
 
@@ -542,9 +540,10 @@ public void run() {
                 try {
                     run.run();
                 } catch (Throwable ex) {
-                    Exceptions.throwIfFatal(ex);
-                    worker.dispose();
-                    throw ExceptionHelper.wrapOrThrow(ex);
+                    // Exceptions.throwIfFatal(ex); nowhere to go
+                    dispose();
+                    RxJavaPlugins.onError(ex);
+                    throw ex;
                 }
             }
         }
@@ -586,7 +585,13 @@ static final class DisposeTask implements Disposable, Runnable, SchedulerRunnabl
         public void run() {
             runner = Thread.currentThread();
             try {
-                decoratedRun.run();
+                try {
+                    decoratedRun.run();
+                } catch (Throwable ex) {
+                    // Exceptions.throwIfFatal(e); nowhere to go
+                    RxJavaPlugins.onError(ex);
+                    throw ex;
+                }
             } finally {
                 dispose();
                 runner = null;

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

@@ -320,6 +320,10 @@ public void run() {
                 }
                 try {
                     actual.run();
+                } catch (Throwable ex) {
+                    // Exceptions.throwIfFatal(ex); nowhere to go
+                    RxJavaPlugins.onError(ex);
+                    throw ex;
                 } finally {
                     lazySet(true);
                 }
@@ -386,7 +390,13 @@ public void run() {
                     thread = Thread.currentThread();
                     if (compareAndSet(READY, RUNNING)) {
                         try {
-                            run.run();
+                            try {
+                                run.run();
+                            } catch (Throwable ex) {
+                                // Exceptions.throwIfFatal(ex); nowhere to go
+                                RxJavaPlugins.onError(ex);
+                                throw ex;
+                            }
                         } finally {
                             thread = null;
                             if (compareAndSet(RUNNING, FINISHED)) {
@@ -463,11 +473,17 @@ public void run() {
             Runnable r = get();
             if (r != null) {
                 try {
-                    r.run();
-                } finally {
-                    lazySet(null);
-                    timed.lazySet(DisposableHelper.DISPOSED);
-                    direct.lazySet(DisposableHelper.DISPOSED);
+                    try {
+                        r.run();
+                    } finally {
+                        lazySet(null);
+                        timed.lazySet(DisposableHelper.DISPOSED);
+                        direct.lazySet(DisposableHelper.DISPOSED);
+                    }
+                } catch (Throwable ex) {
+                    // Exceptions.throwIfFatal(ex); nowhere to go
+                    RxJavaPlugins.onError(ex);
+                    throw ex;
                 }
             }
         }

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

@@ -20,7 +20,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.rxjava3.disposables.Disposable;
-import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.internal.functions.Functions;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 
@@ -54,12 +53,13 @@ public Void call() {
         runner = Thread.currentThread();
         try {
             task.run();
-            setRest(executor.submit(this));
             runner = null;
+            setRest(executor.submit(this));
         } catch (Throwable ex) {
-            Exceptions.throwIfFatal(ex);
+            // Exceptions.throwIfFatal(ex); nowhere to go
             runner = null;
             RxJavaPlugins.onError(ex);
+            throw ex;
         }
         return null;
     }

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,10 +38,11 @@ public void run() {
             runnable.run();
             runner = null;
         } catch (Throwable ex) {
-            Exceptions.throwIfFatal(ex);
+            // Exceptions.throwIfFatal(ex); nowhere to go
             runner = null;
-            lazySet(FINISHED);
+            dispose();
             RxJavaPlugins.onError(ex);
+            throw 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,16 @@ 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);
+            throw ex;
         }
         return null;
     }

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

@@ -66,6 +66,7 @@ public void run() {
             } catch (Throwable e) {
                 // Exceptions.throwIfFatal(e); nowhere to go
                 RxJavaPlugins.onError(e);
+                throw e;
             }
         } finally {
             lazySet(THREAD_INDEX, null);

src/test/java/io/reactivex/rxjava3/core/DisposeTaskTest.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava3.core;
+
+import static org.junit.Assert.fail;
+import static org.testng.Assert.assertTrue;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.Scheduler.DisposeTask;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class DisposeTaskTest extends RxJavaTest {
+
+    @Test
+    public void runnableThrows() throws Throwable {
+        TestHelper.withErrorTracking(errors -> {
+
+            Scheduler.Worker worker = Schedulers.single().createWorker();
+
+            DisposeTask task = new DisposeTask(() -> {
+                throw new TestException();
+            }, worker);
+
+            try {
+                task.run();
+                fail("Should have thrown!");
+            } catch (TestException expected) {
+                // expected
+            }
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+
+            assertTrue(worker.isDisposed());
+        });
+    }
+}

src/test/java/io/reactivex/rxjava3/core/PeriodicDirectTaskTest.java

@@ -0,0 +1,57 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava3.core;
+
+import static org.junit.Assert.fail;
+import static org.testng.Assert.assertTrue;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.Scheduler.PeriodicDirectTask;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class PeriodicDirectTaskTest extends RxJavaTest {
+
+    @Test
+    public void runnableThrows() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            Scheduler.Worker worker = Schedulers.single().createWorker();
+
+            PeriodicDirectTask task = new PeriodicDirectTask(() -> {
+                throw new TestException();
+            }, worker);
+
+            try {
+                task.run();
+                fail("Should have thrown!");
+            } catch (TestException expected) {
+                // expected
+            }
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+
+            assertTrue(worker.isDisposed());
+
+            task.run();
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+}

src/test/java/io/reactivex/rxjava3/internal/schedulers/BooleanRunnableTest.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava3.internal.schedulers;
+
+import static org.junit.Assert.fail;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.RxJavaTest;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.internal.schedulers.ExecutorScheduler.ExecutorWorker.BooleanRunnable;
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class BooleanRunnableTest extends RxJavaTest {
+
+    @Test
+    public void runnableThrows() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            BooleanRunnable task = new BooleanRunnable(() -> {
+                throw new TestException();
+            });
+
+            try {
+                task.run();
+                fail("Should have thrown!");
+            } catch (TestException expected) {
+                // expected
+            }
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+}

src/test/java/io/reactivex/rxjava3/internal/schedulers/InstantPeriodicTaskTest.java

@@ -44,7 +44,12 @@ public void run() {
                 }
             }, exec);
 
-            assertNull(task.call());
+            try {
+                task.call();
+                fail("Should have thrown!");
+            } catch (TestException excepted) {
+                // excepted
+            }
 
             TestHelper.assertUndeliverable(errors, 0, TestException.class);
         } finally {

src/test/java/io/reactivex/rxjava3/internal/schedulers/InterruptibleRunnableTest.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava3.internal.schedulers;
+
+import static org.junit.Assert.fail;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.RxJavaTest;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.internal.schedulers.ExecutorScheduler.ExecutorWorker.InterruptibleRunnable;
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class InterruptibleRunnableTest extends RxJavaTest {
+
+    @Test
+    public void runnableThrows() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            InterruptibleRunnable task = new InterruptibleRunnable(() -> {
+                throw new TestException();
+            }, null);
+
+            try {
+                task.run();
+                fail("Should have thrown!");
+            } catch (TestException expected) {
+                // expected
+            }
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+}