Pub/Sub has no way to track errors from the subscriber thread.

[theacodes]

Presently if you subscribe and an error occurs in the consumer's thread there's no way to respond to the error on the main thread.

It just logs the exception and hangs:

Exception in thread Consumer helper: consume bidirectional stream:
Traceback (most recent call last):
 File "/Users/jonwayne/.pyenv/versions/3.6.0/lib/python3.6/threading.py", line 916, in _bootstrap_inner
 self.run()
 File "/Users/jonwayne/.pyenv/versions/3.6.0/lib/python3.6/threading.py", line 864, in run
 self._target(*self._args, **self._kwargs)
 File "/Users/jonwayne/workspace/python-docs-samples/pubsub/cloud-client/env/lib/python3.6/site-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 248, in _blocking_consume
 self._policy.on_exception(exc)
 File "/Users/jonwayne/workspace/python-docs-samples/pubsub/cloud-client/env/lib/python3.6/site-packages/google/cloud/pubsub_v1/subscriber/policy/thread.py", line 135, in on_exception
 raise exception
 File "/Users/jonwayne/workspace/python-docs-samples/pubsub/cloud-client/env/lib/python3.6/site-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 234, in _blocking_consume
 for response in response_generator:
 File "/Users/jonwayne/workspace/python-docs-samples/pubsub/cloud-client/env/lib/python3.6/site-packages/grpc/_channel.py", line 363, in __next__
 return self._next()
 File "/Users/jonwayne/workspace/python-docs-samples/pubsub/cloud-client/env/lib/python3.6/site-packages/grpc/_channel.py", line 357, in _next
 raise self
grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.NOT_FOUND, Resource not found (resource=my-subs).)>

It seems that consumer.open should return a future. This future could be used to both block the main thread to consume message and pass through exceptions.

So the current sample:

def receive_messages(project, subscription_name):
 """Receives messages from a pull subscription."""
 subscriber = pubsub_v1.SubscriberClient()
 subscription_path = subscriber.subscription_path(
 project, subscription_name)

 def callback(message):
 print('Received message: {}'.format(message))
 message.ack()

 subscriber.subscribe(subscription_path, callback=callback)

 # The subscriber is non-blocking, so we must keep the main thread from
 # exiting to allow it to process messages in the background.
 print('Listening for messages on {}'.format(subscription_path))
 while True:
 time.sleep(60)

Could instead be:

def receive_messages(project, subscription_name):
 """Receives messages from a pull subscription."""
 subscriber = pubsub_v1.SubscriberClient()
 subscription_path = subscriber.subscription_path(
 project, subscription_name)

 def callback(message):
 print('Received message: {}'.format(message))
 message.ack()

 future = subscriber.subscribe(subscription_path, callback=callback)

 print('Listening for messages on {}'.format(subscription_path))
 # The subscriber is non-blocking but returns a future that allows us
 # to block the main thread and allow messages to process in the
 # background. This effectively blocks forever unless an error
 # occurs in the subscriber thread in which case it'll be
 # re-raised here.
 future.result()

[lukesneeringer]

I think I agree with this but it is worth asking: Is there any situation where there are multiple exceptions? Futures can only have single resolutions.

I do think it is fine to have a future that never has a result (or only gets one if you explicitly close), but I am operating under the assumption that when the exception occurs, we will cause that thread to exit and set the exception on the future.

[theacodes]

I think that any exception that bubbles to the top of the consumer thread is a thread ending exception, so there should be only one. The rest of your comment matches my understanding.

…

On Mon, Aug 28, 2017, 10:07 PM Luke Sneeringer ***@***.***> wrote: I *think* I agree with this but it is worth asking: Is there any situation where there are multiple exceptions? Futures can only have single resolutions. I do think it is fine to have a future that never has a result (or only gets one if you explicitly close), but I am operating under the assumption that when the exception occurs, we will cause that thread to exit and set the exception on the future. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#3888 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAPUczQJq3yTWzMB1tJG97X3U-w8WS_cks5sc5yLgaJpZM4PE4py> .

[duilio]

A workaround would be to override the default Policy used by the Client. For instance:

class Custom(Policy):
 def on_exception(self, exception):
 try:
 return super(Custom, self).on_exception(exception)
 except:
 # handle consumer exceptions here, for instance reject a promise or
 # set threading.Event to communicate the failure to the main thread.
 raise

[cgthayer]

Having a future to check on would be a whole lot cleaner probably. At least it would be clear that control was being passed around, whereas having the on_exception() is vague and requires understanding the internals some.

We've had to do something similar because we see this error which never seems to recover (running in google container engine):

grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.UNAVAILABLE, OS Error)>

I'm inclined to believe that a connection broke or shutdown, causing a socket to trigger an "OS Error".
This is not the only error we see, but since they show up sporadically it's hard to know which ones wedge us and which are properly handled. For example DEADLINE_EXCEEDED is "normal" and processing goes on.

To work around the wedging, our code does something like this. We set a variable to indicate to the outer loop that the client might be bad.

class OurPolicy(thread.Policy):
 """ 
 We occasionally see errors that google code doesn't recover from, so we 
 set a flag that let's the outer thread respond by restarting the client. 
 
 grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.UNAVAILABLE, OS Error)> 
 
 """
 _exception_caught = None

 def __init__(self, *args, **kws):
 logger.info(f'Initializing our PubSub Policy Wrapper') # noqa 
 return super(OurPolicy, self).__init__(*args, **kws)

 def on_exception(self, exc):
 # If this is DEADLINE_EXCEEDED, then we want to retry by returning None instead of raise-ing 
 deadline_exceeded = grpc.StatusCode.DEADLINE_EXCEEDED
 code_value = getattr(exc, 'code', lambda: None)()
 logger.error(f'Caught Exception in PubSub Policy Wrapper: code={code_value} exc={exc}')
 if code_value == deadline_exceeded:
 return
 OurPolicy._exception_caught = exc
 # will just raise exc 
 return super(OurPolicy, self).on_exception(exc)


def get_publisher_client():
 return pubsub_v1.PublisherClient(
 batch_settings=types.BatchSettings(max_latency=0.001, max_messages=1))


def get_subscriber_client():
 return pubsub_v1.SubscriberClient(policy_class=OurPolicy)

then our listener can tell when we've hit an exception that would wedge our server, and instead it restarts the client:

class Worker(object):
 [..lots deleted..]

 def listen(self):
 logger.info('Listening, press Ctrl+C to exit.')
 subscriber = get_subscriber_client()
 subscription_path = subscriber.subscription_path(NOTIFY_PROJECT, self.subscription_name)
 while True:
 if not self.sub_client:
 logger.warning('Restarting pubsub subscriber client')
 self.sub_client = get_subscriber_client()
 self.sub_client.subscribe(subscription_path, callback=self.message_callback)
 try:
 # The subscriber is non-blocking, so we must keep the main thread from 
 # exiting to allow it to process messages in the background. 
 logger.info(f'Listening for messages on topic {self.topic_name} @ {subscription_path}')
 while True:
 time.sleep(1)
 if OurPolicy._exception_caught:
 exc = OurPolicy._exception_caught
 OurPolicy._exception_caught = None
 raise exc
 except KeyboardInterrupt:
 logger.info('Stopped listening for tasks.')
 break
 except Exception as e:
 logger.exception(e)
 self.sub_client = None
 from raven.contrib.django.raven_compat.models import client
 client.captureException()

That said, I'm not sure if we should add some mutexes here and be more thread-safe. Also this may have other issues since it's unclear if the old client ever shuts down cleanly. There's nothing in the API docs about it, and python can be lazy about this kind of resource cleanup, especially since worker threads may linger with a reference to the old client object.

https://googlecloudplatform.github.io/google-cloud-python/latest/pubsub/publisher/api/client.html