Pubsub consumers regularly fill with InvalidArgument

[sachin-shetty]

1. Specify the API at the beginning of the title (for example, "BigQuery: ...")

Google Pubsub Python Client

2. OS type and version

Linux: 3.10.0-693.5.2.el7.x86_64 #1 SMP Fri Oct 20 20:32:50 UTC 2017 x86_64 x86_64 x86_64 GNU/Linux

3. Python version and virtual environment information python --version

Python 2.7.5

4. google-cloud-python version pip show google-cloud, pip show google-<service> or pip freeze

Name: google-cloud-pubsub
Version: 0.30.1
Summary: Python Client for Google Cloud Pub/Sub
Home-page: https://github.com/GoogleCloudPlatform/google-cloud-python
Author: Google Cloud Platform
Author-email: googleapis-packages@google.com
License: Apache 2.0
Location: /usr/local/lib/python2.7/dist-packages
Requires: google-api-core, psutil, grpc-google-iam-v1, google-auth

5. Stacktrace if available

    future.result()
  File "/usr/lib/python2.7/site-packages/google/cloud/pubsub_v1/futures.py", line 98, in result
    raise err
InvalidArgument: 400 Request contains an invalid argument.

6. Steps to reproduce
7. Code example

future = subscription.open(callback)
future.result()

After a while consumer stops receiving messages, and then I see the above error in the log.

    future.result()
  File "/usr/lib/python2.7/site-packages/google/cloud/pubsub_v1/futures.py", line 98, in result
    raise err
InvalidArgument: 400 Request contains an invalid argument.

Sometimes the error goes away on closing and re-opening the subscription, sometimes it comes back sooner.

Is there any workaround to monitor the state of the pull subscription and then gracefully restart it?

[sachin-shetty]

Got a different error now

Code:

            future = subscription.open(self.callback)
            try:
                logger.info("Waiting for messages on %s", self.subscriber_name )
                future.result()
            except Exception as ex:
                logging.exception("Error in blocked result")
                subscription.close()

Stack trace

Error in blocked result
Traceback (most recent call last):
  File "video_transcode.py", line 337, in run
    future.result()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/futures.py", line 103, in result
    raise err
Unknown: None Stream removed
Error in blocked result
Traceback (most recent call last):
  File "video_transcode.py", line 337, in run
    future.result()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/futures.py", line 103, in result
    raise err
Unknown: None Stream removed
Exception in thread Thread-9:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "video_transcode.py", line 341, in run
    subscription.close()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/policy/thread.py", line 174, in close
    self._consumer.stop_consuming()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 441, in stop_consuming
    thread = self._stop_no_join()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 426, in _stop_no_join
    _LOGGER.debug('Stopping helper thread %s', self._consumer_thread.name)
AttributeError: 'NoneType' object has no attribute 'name'
Exception in thread Thread-8:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "video_transcode.py", line 341, in run
    subscription.close()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/policy/thread.py", line 174, in close
    self._consumer.stop_consuming()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 441, in stop_consuming
    thread = self._stop_no_join()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 426, in _stop_no_join
    _LOGGER.debug('Stopping helper thread %s', self._consumer_thread.name)
AttributeError: 'NoneType' object has no attribute 'name'

[sachin-shetty]

Some more info, I am now getting Unknown: None Stream removed from future.result() in about 15 minutes regularly.

Traceback (most recent call last):
  File "video_transcode.py", line 338, in run
    try:
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/futures.py", line 103, in result
    raise err
Unknown: None Stream removed

The subsequent error on subscriber.stop could be due to a custom thread handler I have, I guess the consumer still looks for the built in one.

Exception in thread Thread-8:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "video_transcode.py", line 341, in run
    subscription.close()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/policy/thread.py", line 174, in close
    self._consumer.stop_consuming()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 441, in stop_consuming
    thread = self._stop_no_join()
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 426, in _stop_no_join
    _LOGGER.debug('Stopping helper thread %s', self._consumer_thread.name)
AttributeError: 'NoneType' object has no attribute 'name'

[theacodes]

@sachin-shetty Can you give us a bit more details? What did you change between these comments to get these different errors?

[theacodes]

I can not reproduce this error and it's been 2 weeks since the last reply. I'm going to close this for now, but happy to re-open if it's still an issue.