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.

google-cloud-python/pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

Lines 533 to 547 in 52b54aa

	def _on_response(self, response):
	"""Process all received Pub/Sub messages.
	For each message, send a modified acknowledgment request to the
	server. This prevents expiration of the message due to buffering by
	gRPC or proxy/firewall. This makes the server and client expiration
	timer closer to each other thus preventing the message being
	redelivered multiple times.
	After the messages have all had their ack deadline updated, execute
	the callback for each message using the executor.
	"""
	_LOGGER.debug(
	"Processing %s received message(s), currenty on hold %s (bytes %s).",
	len(response.received_messages),

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.