RxJava2 "fatal" exceptions thrown by subscribers are swallowed by rxFlowable

[zach-klippenstein]

This issue occurs under the following conditions:

1. RxJava2 Flowable created using the rxFlowable coroutine builder. Note that rxObservable is not affected.
2. The resulting Flowable is multicasted using either replay or publish. Method of connection doesn't matter (can use either autoConnect, refCount, or connect manually with connect).
3. Flowable is subscribed with single-argument subscribe, taking a Consumer, and the consumer throws a "fatal" exception, as defined by RxJava2's Exceptions.throwIfFatal method. Note that non-fatal exceptions get wrapped in OnErrorNotImplemented exceptions and re-thrown, as expected.

Expected behavior: The exception gets thrown as an OnErrorNotImplementedException, or at least reported to the default uncaught exception handler.
Observed behavior: The exception is swallowed without being reported in any way.

The context for the rxFlowable builder can be empty (GlobalScope + EmptyCoroutineContext), or use the Unconfined dispatcher, the issue occurs in both cases.

I have a Gradle project that reproduces the issue in the simplest form I could whittle it down to: https://github.com/zach-klippenstein/coroutines-rxflowable-bug-repo (see Main.kt).