PubSub: Making `thread.Policy.on_exception` more robust. #4444

dhermes · 2017-11-23T00:38:40Z

NOTE: I am NOT comfortable with this as a fix, it seems WAAAAAY too bolted on.

~~I'm surprised the retry strategy doesn't come into play here~~
~~There should a less hard-coded way to map to the idempotent retry codes~~
~~The method should have a straightforward way to determine the gRPC status code from the exception~~
RuntimeError: set_exception can only be called once occurs if I don't filter out the UNAVAILABLE (e.g.). This seems like a bug in a different code path than is covered here.

Adding special handling for API core exceptions
Retrying on both types of idempotent error
Also doing a "drive-by" hygiene fix changing a global from logger to _LOGGER

Towards #4234.

pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py

+            code = exception.grpc_status_code
+        else:
+            code = getattr(exception, 'code', None)
+            if callable(code):


pubsub/tests/unit/pubsub_v1/subscriber/test_policy_thread.py

+    details = 'Bad thing happened. Time out, go sit in the corner.'
+    exc_state = grpc._channel._RPCState(
+        (), None, trailing, status_code, details)
+    exc = grpc._channel._Rendezvous(exc_state, None, None, None)


theacodes · 2017-11-27T20:10:12Z

I'm surprised the retry strategy doesn't come into play here

It's because this happens during streaming. api_core can't know the semantics for restarting the stream. In this case, it's simple, in others (such as spanner) it's not.

pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py

-        if getattr(exception, 'code', lambda: None)() == deadline_exceeded:
+        if isinstance(exception, exceptions.GoogleAPICallError):
+            code = exception.grpc_status_code
+        else:


pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py

+
+        # If this is in the list of idempotent exceptions DEADLINE_EXCEEDED,
+        # then we want to retry. That entails just returning None.
+        if code in _IDEMPOTENT_RETRY_CODES:


- Adding special handling for API core exceptions - Retrying on both types of idempotent error - Also doing a "drive-by" hygiene fix changing a global from `logger` to `_LOGGER` Towards googleapis#4234.

Also removing the `logger` -> `_LOGGER` change to make the PR easier to review.

dhermes · 2017-11-27T21:50:19Z

So I have updated and have a better understanding thanks to chats with @jonparrott and @lukesneeringer.

We only expect exceptions from api_core here. This is even true for iterators.
I added BasePolicy._IDEMPOTENT_RETRIES and just added a unit test to sanity check that it matches the config. (I think this is preferred to an import time penalty for every user ever.)
Using isinstance(exception, _IDEMPOTENT_RETRIES), all of the "branch to check for Rendezvous? is code callable? huh?" issues are gone.

I'm still not sure why set_exception gets called twice (crossref #4380). I'm going to try to reproduce this by removing exceptions.ServiceUnavailable from _IDEMPOTENT_RETRIES and then running the code in my gist.

I would like to continue that exploration but still merge this PR. SGTY @jonparrott and @lukesneeringer?

pubsub/google/cloud/pubsub_v1/subscriber/policy/base.py

@@ -65,6 +66,10 @@ class BasePolicy(object):
    """

    _managed_ack_ids = None
+    _IDEMPOTENT_RETRIES = (


pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py

-        if getattr(exception, 'code', lambda: None)() == deadline_exceeded:
+        # If this is in the list of idempotent exceptions, then we want to
+        # retry. That entails just returning None.
+        if isinstance(exception, self._IDEMPOTENT_RETRIES):


theacodes · 2017-11-27T21:57:28Z

I would like to continue that exploration but still merge this PR.

dhermes · 2017-11-27T22:12:11Z

Well I was able to reproduce

Traceback (most recent call last):
  File "${HOME}/.pyenv/versions/3.6.3/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "${HOME}/.pyenv/versions/3.6.3/lib/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
  File ".../google/cloud/pubsub_v1/subscriber/_consumer.py", line 249, in _blocking_consume
    self._policy.on_exception(exc)
  File ".../google/cloud/pubsub_v1/subscriber/policy/thread.py", line 162, in on_exception
    self._future.set_exception(exception)
  File ".../google/cloud/pubsub_v1/futures.py", line 159, in set_exception
    raise RuntimeError('set_exception can only be called once.')
RuntimeError: set_exception can only be called once.

on my very first run with the hack

from google.cloud.pubsub_v1.subscriber.policy import base
# Monkey-patch to intentionally miss errors.
base.BasePolicy._RETRYABLE_STREAM_ERRORS = ()

dhermes added the api: pubsub Issues related to the Pub/Sub API. label Nov 23, 2017

dhermes assigned lukesneeringer Nov 23, 2017

dhermes requested a review from lukesneeringer as a code owner November 23, 2017 00:38

googlebot added the cla: yes This human has signed the Contributor License Agreement. label Nov 23, 2017

chemelnucfin changed the title ~~Making thread.Policy.on_exception more robust.~~ PubSub: Making thread.Policy.on_exception more robust. Nov 23, 2017

dhermes mentioned this pull request Nov 27, 2017

Pub/Sub Subscriber does not catch & retry UNAVAILABLE errors #4234

Closed

dhermes force-pushed the towards-4234 branch from 4244392 to ae0b073 Compare November 27, 2017 19:06

theacodes reviewed Nov 27, 2017

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py Outdated

if getattr(exception, 'code', lambda: None)() == deadline_exceeded:

if isinstance(exception, exceptions.GoogleAPICallError):

code = exception.grpc_status_code

else:

This comment was marked as spam.

Sign in to view

theacodes reviewed Nov 27, 2017

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py Outdated

# If this is in the list of idempotent exceptions DEADLINE_EXCEEDED,

# then we want to retry. That entails just returning None.

if code in _IDEMPOTENT_RETRY_CODES:

This comment was marked as spam.

Sign in to view

Making thread.Policy.on_exception more robust.

7b9c04c

- Adding special handling for API core exceptions - Retrying on both types of idempotent error - Also doing a "drive-by" hygiene fix changing a global from `logger` to `_LOGGER` Towards googleapis#4234.

dhermes mentioned this pull request Nov 27, 2017

Fix RuntimeError: set_exception can only be called once #4380

Closed

Only catching exceptions from api_core.

30a949f

Also removing the `logger` -> `_LOGGER` change to make the PR easier to review.

dhermes force-pushed the towards-4234 branch from ae0b073 to 30a949f Compare November 27, 2017 21:44

theacodes approved these changes Nov 27, 2017

View reviewed changes

Renaming _IDEMPOTENT_RETRIES -> _RETRYABLE_STREAM_ERRORS.

027ded4

dhermes merged commit 83865c2 into googleapis:master Nov 27, 2017

dhermes deleted the towards-4234 branch November 27, 2017 22:10

dhermes mentioned this pull request Dec 15, 2017

PubSub with Python 3.5 spawns threads that don't die, eventually bringing the process to consume all CPU available #4563

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PubSub: Making `thread.Policy.on_exception` more robust. #4444

PubSub: Making `thread.Policy.on_exception` more robust. #4444

dhermes commented Nov 23, 2017 •

edited

Loading

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

theacodes commented Nov 27, 2017

This comment was marked as spam.

This comment was marked as spam.

dhermes commented Nov 27, 2017

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

theacodes commented Nov 27, 2017

dhermes commented Nov 27, 2017

PubSub: Making thread.Policy.on_exception more robust. #4444

PubSub: Making thread.Policy.on_exception more robust. #4444

Conversation

dhermes commented Nov 23, 2017 • edited Loading

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

theacodes commented Nov 27, 2017

This comment was marked as spam.

This comment was marked as spam.

dhermes commented Nov 27, 2017

This comment was marked as spam.

This comment was marked as spam.

This comment was marked as spam.

theacodes commented Nov 27, 2017

dhermes commented Nov 27, 2017

PubSub: Making `thread.Policy.on_exception` more robust. #4444

PubSub: Making `thread.Policy.on_exception` more robust. #4444

dhermes commented Nov 23, 2017 •

edited

Loading