Google-cloud-python: PubSub: AttributeError: 'NoneType' object has no attribute 'received_messages'

[briancurtin]

Environment details

OS: OS X and Linux
Python 3.8.1rc1
google-cloud-pubsub==1.1.0
PubSub Emulator running out of the google/cloud-sdk:latest docker image

Steps to reproduce

I'm working on getting an isolated reproduction of this, but at current it's "run my integration tests against the PubSub emulator" (closed source, unfortunately) and see that on 1.1.0 the stack trace listed below gets written to stdout/stderr as it's seemingly occurring in another thread (not in my logs), and the test suite continues on normally and eventually passes. When pinned on 1.0.2, the stack trace below does not occur.

For whatever it's worth, our code using PubSub via a streaming pull future has not changed recently.

I get that this isn't very helpful by itself but I wanted to get report this sooner than later, instead of just pinning to 1.0.2 and forgetting about it.

Stack trace

ERROR:google.api_core.bidi:Thread-ConsumeBidirectionalStream caught unexpected exception 'NoneType' object has no attribute 'received_messages' and will exit.
Traceback (most recent call last):
  File "/Users/briancurtin/elastic/cloud/python-services-v3/.tox/integration/lib/python3.8/site-packages/google/api_core/bidi.py", line 657, in _thread_main
    self._on_response(response)
  File "/Users/briancurtin/elastic/cloud/python-services-v3/.tox/integration/lib/python3.8/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 547, in _on_response
    len(response.received_messages),
AttributeError: 'NoneType' object has no attribute 'received_messages'

[kpurdon]

While this won't replicate every time for me I can get it to trigger this exception every 5th or so run. It's odd to manually close the underlying channel, but might be useful to track the issue down. Of note when this occurs for me in normal code i'm not manually closing the channel.

from google.cloud import pubsub_v1


def callback(message):
    print(message)


if __name__ == "__main__":
    sc = pubsub_v1.SubscriberClient()
    future = sc.subscribe(sc.subscription_path("development", "pstest"), callback)
    sc.api.transport.channel.close()
    future.result()

$ python main.py
Thread-ConsumeBidirectionalStream caught unexpected exception 'NoneType' object has no attribute 'received_messages' and will exit.
Traceback (most recent call last):
  File "/tmp/pstest/.direnv/python-3.7.5/lib/python3.7/site-packages/google/api_core/bidi.py", line 657, in _thread_main
    self._on_response(response)
  File "/tmp/pstest/.direnv/python-3.7.5/lib/python3.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 547, in _on_response
    len(response.received_messages),
AttributeError: 'NoneType' object has no attribute 'received_messages'
Traceback (most recent call last):
  File "main.py", line 13, in <module>
    future.result()
  File "/tmp/pstest/.direnv/python-3.7.5/lib/python3.7/site-packages/google/cloud/pubsub_v1/futures.py", line 105, in result
    raise err
google.api_core.exceptions.Cancelled: 499 Channel closed!

Running with the same configuration as the initial report.

[plamut]

I was able to reproduce the issue by running a subscriber that pulls the messages using a streaming pull, and then cancelling the future with a keyboard interrupt. In fact, the reproduction rate is 100% on my machine with my own test script.

future = subscriber.subscribe(subscription_name, callback)
try:
    future.result()
except KeyboardInterrupt:
    future.cancel()

Log output:

2019-12-16 14:15:13,682 ERROR Thread-ConsumeBidirectionalStream: Thread-ConsumeBidirectionalStream caught unexpected exception 'NoneType' object has no attribute 'received_messages' and will exit.
Traceback (most recent call last):
  File "/home/peter/workspace/google-cloud-python/api_core/google/api_core/bidi.py", line 657, in _thread_main
    self._on_response(response)
  File "/home/peter/workspace/google-cloud-python/pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 547, in _on_response
    len(response.received_messages),
AttributeError: 'NoneType' object has no attribute 'received_messages'

It appears that upon cancelling, the internal "on message receive" callback is passed a StreamingPullResponse instance with an empty received_messages attribute (its value is None rather than an empty message list). None value instead of a StreamingPullResponse instance.
https://github.com/googleapis/google-cloud-python/blob/52b54aaa93f47c772de14706ba5bf300701aa8bf/pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py#L533-L547

Should be straightforward to add logic to handle this case.

@anguillanneuf I can fix this pretty soon, just please let me know if you have already started working on this to not duplicate the work, thanks!

Update: Actually found some extra time even today, and fixed it myself. Opened #9982.