PubSub: Prevent unhandled background error on SPM shutdown

[plamut]

Closes #6130.

This PR fixes errors in background threads that might happen on user-initiated streaming pull shutdown, i.e. cancelling the future returned by the subscriber.subscribe() call.

How to test

Steps to reproduce:

• Follow the steps in the issue description.
• Preferably set the log level of the test script to logging.DEBUG. FWIW, I used the following config during development:

import logging

log_format = (
    "%(levelname)-8s [%(asctime)s] %(threadName)-33s "
    "[%(name)s] [%(filename)s:%(lineno)d][%(funcName)s] %(message)s"
)
logging.basicConfig(level=logging.DEBUG, format=log_format)

Actual result (before the fix):
An error can be observed in Thread-CallbackRequestDispatcher (the one listed under "Stacktrace 2" in the issue description). The issue is reproducible with high consistency:

AttributeError: 'NoneType' object has no attribute 'remove'

NOTE: The error listed under "Stacktrace 1" is not reproducible anymore, it has been fixed quite some time ago.

Expected result (after the fix):
The StreamingPullManager gets shut down cleanly, i.e. without the error reported. There might be a stacktrace of the Cancelled exception produced by api_core.bidi, but that is a harmless "expected exception":

Thread-ConsumeBidirectionalStream caught error 499 Locally cancelled by application! and will exit. Generally this is due to the RPC itself being cancelled and the error will be surfaced to the calling code
...
google.api_core.exceptions.Cancelled: 499 Locally cancelled by application!

[plamut]

The main problem here is that the leaser and dispatcher use each other's methods, thus cannot be shut down in a clear linear sequence (and breaking this inter-dependency would likely require substantial refactoring).

It's also tricky to synchronize the manager shutdown with any remaining work that dispatcher might have left in its queue (such as pending ACK requests). The thread that initiated the manager shutdown waits on the dispatcher thread to terminate, thus if the latter tries to sync with the manager shutdown through a shared lock, a deadlock can happen.

The current fix uses the fact that once in the shutdown mode, several operations on the manager have no effect (maybe_resume_consumer() / maybe_pause_consumer()), as do not the add() and remove() operations on a stopped leaser.

I still need to think of a meaningful yet understandable test to properly cover this.

[tseaver]

Could we delay setting the attributes to None until the very end?

[plamut]

@tseaver Definitely, that's exactly the fix that I am just about to submit. Much easier and cleaner than trying to refactor the streaming pull and trying to break the cyclic dependency between the dispatcher and the leaser.