Streaming pull shutdown hangs indefinitely on error in callback

#### Environment details

  - OS type and version: 4.19.121-linuxkit
  - Python version: 3.8.9
  - pip version: 21.0.1
  - `google-cloud-pubsub` version: 2.4.1

#### Steps to reproduce

  1. Use the StreamingPull subscriber
  2. Throw an exception in the callback function
  3. Will in some cases hang indefinitely
 
When an exception occurs in the callback, the subscriber will hang indefinitely in some cases (about 1 in 5). It seems the queue returns `None` instead of `queue.Empty` here: https://github.com/googleapis/python-pubsub/blob/58120185d6712ae5f84d41ff6e41dd706a6463ca/google/cloud/pubsub_v1/subscriber/scheduler.py#L142-L147 

This happens both with `await_callbacks_on_shutdown=True` and `await_callbacks_on_shutdown=False`.

#### Code example

```python
from concurrent.futures import TimeoutError
from google.cloud.pubsub import SubscriberClient

project_id = "your-project-id"
subscription_id = "your-subscription-id"

subscriber = SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def callback(message):
    raise Exception()

    message.ack()

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path}..\n")

with subscriber:
    try:
        streaming_pull_future.result()
    except TimeoutError:
        streaming_pull_future.cancel()
```

#### Stack trace
```
Exception in thread Thread-OnRpcTerminated:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/usr/local/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.8/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 569, in _shutdown
    dropped_messages = self._scheduler.shutdown(
  File "/usr/local/lib/python3.8/site-packages/google/cloud/pubsub_v1/subscriber/scheduler.py", line 145, in shutdown
    dropped_messages.append(work_item.args[0])
AttributeError: 'NoneType' object has no attribute 'args'
```


	try:
	while True:
	work_item = self._executor._work_queue.get(block=False)
	dropped_messages.append(work_item.args[0])
	except queue.Empty:
	pass

Navigation Menu

Uh oh!

Streaming pull shutdown hangs indefinitely on error in callback #394

Environment details

Steps to reproduce

Code example

Stack trace

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Uh oh!

Streaming pull shutdown hangs indefinitely on error in callback #394

Description

Environment details

Steps to reproduce

Code example

Stack trace

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions