This was discovered in Pub/Sub, whereby futures dependent on this class hang if not set with a timeout when a throw occurs in BackgroundConsumer's _thread_main.

When we open a BackgroundConsumer, if something within the try block, e.g. open, throws, then the exception clauses log but do not close the underlying stream or thread.

I believe this will be fixed by adding allowing an optional parameter on_fatal_exception with which the consumer can opt to perform behavior with the fatal exception. With this, we can fail futures using the original exception that caused the RPC lifecycle to fail.

I will be submitting a PR to address this.