Undeliverable error handling logic for Completable operators (#6442)

* Error handle on Completable.fromAction with RxJavaPlugins

* Error handle on Completable.fromRunnable with RxJavaPlugins

* Added error handling java docs section to Completable.fromRunnable

Author: BharathMG

Diff for: src/main/java/io/reactivex/Completable.java

@@ -508,6 +508,13 @@ public static <T> Completable fromMaybe(final MaybeSource<T> maybe) {
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code fromRunnable} does not operate by default on a particular {@link Scheduler}.</dd>
+     *  <dt><b>Error handling:</b></dt>
+     *  <dd> If the {@link Runnable} throws an exception, the respective {@link Throwable} is
+     *  delivered to the downstream via {@link CompletableObserver#onError(Throwable)},
+     *  except when the downstream has disposed this {@code Completable} source.
+     *  In this latter case, the {@code Throwable} is delivered to the global error handler via
+     *  {@link RxJavaPlugins#onError(Throwable)} as an {@link io.reactivex.exceptions.UndeliverableException UndeliverableException}.
+     *  </dd>
      * </dl>
      * @param run the runnable to run for each subscriber
      * @return the new Completable instance

Diff for: src/main/java/io/reactivex/internal/operators/completable/CompletableFromAction.java

@@ -17,6 +17,7 @@
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Action;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class CompletableFromAction extends Completable {
 
@@ -36,6 +37,8 @@ protected void subscribeActual(CompletableObserver observer) {
             Exceptions.throwIfFatal(e);
             if (!d.isDisposed()) {
                 observer.onError(e);
+            } else {
+                RxJavaPlugins.onError(e);
             }
             return;
         }

Diff for: src/main/java/io/reactivex/internal/operators/completable/CompletableFromRunnable.java

@@ -18,6 +18,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.disposables.Disposables;
 import io.reactivex.exceptions.Exceptions;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class CompletableFromRunnable extends Completable {
 
@@ -37,6 +38,8 @@ protected void subscribeActual(CompletableObserver observer) {
             Exceptions.throwIfFatal(e);
             if (!d.isDisposed()) {
                 observer.onError(e);
+            } else {
+                RxJavaPlugins.onError(e);
             }
             return;
         }