Closes#7709.

If a gRPC channel is inTRANSIENT_FAILURE state for too long, the retry timeout configured insubscriber client config kicks in, and aRetryError is raised in a background thread, but the client keeps running, and the error is not propagated to the top level code.

This PR makes sure that the following happens:

• The streaming pull manager shutdown is triggered, shutting down all background threads.
• The reason for the shutdown (RetryError) is propagated to the main thread that awaitsfuture.result(), allowing the user code a chance to catch the error and react to it.

How to test

I was not able to reproduce the actual error users reported in a real setup (a sample pubsub app deployed to K8s), but figured out what isprobably happening and faked the error.

Steps to reproduce:

• Patch the_end_unary_response_blocking() function of thegrpc dependency in your local Python environment, example:

--- /home/peter/workspace/google-cloud-python/venv-3.6/lib/python3.6/site-packages/grpc/_channel.py     2019-04-23 17:01:39.282064676 +0200+++ /home/peter/workspace/google-cloud-python/venv-3.6/lib/python3.6/site-packages/grpc/_channel.py     2019-04-25 15:49:05.220317794 +0200@@ -456,6 +456,11 @@   def _end_unary_response_blocking(state, call, with_call, deadline):+    state.code = grpc.StatusCode.UNAVAILABLE+    state.details = "channel is in **fake** TRANSIENT_FAILURE state"+    state.debug_error_string = (+        "transient failure is faked during a fixed time window in an hour"+    )     if state.code is grpc.StatusCode.OK:         if with_call:             rendezvous = _Rendezvous(state, call, None, deadline)

• Adjust thetotal_timeout_millis setting insubscriber client config to 10 (seconds... in order to not wait for too long)
• Run a sample subscriber script (example). Prerequisite: having a Google service account configured for the subscriber client, and an active subscription to a topic.

Actual result (before the fix):
ARetryError occurs in the background after ~10 seconds, some of the threads exit, but the subscriber client keeps running, and the error isnot propagated to the main thread (the future returned by thesubscribe() method is not resolved)

Expected result (after the fix):
Everything gets shut down cleanly, andRetryError is propagated to and raised in the main code.