Publish future blocks indefinitely if API raises exception #4019
Description
- OS type and version: macOS 10.12.6
- Python version and virtual environment information
python --version: Python 3.4.6 - google-cloud-python version(s):
google-cloud-core==0.27.1
google-cloud-pubsub==0.28.3
dill==0.2.7.1
future==0.16.0
google-auth==1.1.0
google-gax==0.15.14
googleapis-common-protos==1.5.2
grpc-google-iam-v1==0.11.3
grpcio==1.6.0
oauth2client==3.0.0
ply==3.8
protobuf==3.4.0
psutil==5.3.1
- Stacktrace if available
>>> Exception in thread Thread-7:
Traceback (most recent call last):
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/gax/retry.py", line 121, in inner
return to_call(*args)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/gax/retry.py", line 68, in inner
return a_func(*updated_args, **kwargs)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/grpc/_channel.py", line 492, in __call__
return _end_unary_response_blocking(state, call, False, deadline)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/grpc/_channel.py", line 440, in _end_unary_response_blocking
raise _Rendezvous(state, None, None, deadline)
grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.NOT_FOUND, Requested project not found or user does not have access to it (project=not-a-cloud-259388BD5DF4). Make sure to specify the unique project identifier and not the Google Cloud Console display name.)>
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Users/schultzm/.pyenv/versions/3.4.6/lib/python3.4/threading.py", line 911, in _bootstrap_inner
self.run()
File "/Users/schultzm/.pyenv/versions/3.4.6/lib/python3.4/threading.py", line 859, in run
self._target(*self._args, **self._kwargs)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/cloud/pubsub_v1/publisher/batch/thread.py", line 212, in monitor
return self._commit()
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/cloud/pubsub_v1/publisher/batch/thread.py", line 173, in _commit
self.messages,
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/cloud/gapic/pubsub/v1/publisher_client.py", line 314, in publish
return self._publish(request, options)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/gax/api_callable.py", line 452, in inner
return api_caller(api_call, this_settings, request)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/gax/api_callable.py", line 438, in base_caller
return api_call(*args)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/gax/api_callable.py", line 376, in inner
return a_func(*args, **kwargs)
File "/Users/schultzm/.virtualenvs/onsite/lib/python3.4/site-packages/google/gax/retry.py", line 127, in inner
' classified as transient', exception)
google.gax.errors.RetryError: RetryError(Exception occurred in retry method that was not classified as transient, caused by <_Rendezvous of RPC that terminated with (StatusCode.NOT_FOUND, Requested project not found or user does not have access to it (project=not-a-cloud-259388BD5DF4). Make sure to specify the unique project identifier and not the Google Cloud Console display name.)>)
- Steps to reproduce
Attempt to publish to a unknown project/topic (or perhaps a topic you don't have permissions for).
- Code example
from json import dumps
from google.cloud import pubsub_v1
publisher = pubsub_v1.PublisherClient()
topic = 'projects/not-a-cloud-259388BD5DF4/topics/not-a-topic'
data = {'toss': 'exception'}
payload = dumps(data).encode('utf8')
future = publisher.publish(topic, payload)
At this point you'll have a future object, but the Batch thread that is going to send the message is already working on it and tries to call the API. It can throw an exception that is uncaught which will cause the thread to exit. The future that is returned will never get updated and the user has no way to access that exception.
I think just throwing that code in a try/except and running set_exception on all the self._futures would suffice?