Repository navigation
Pub/Sub Subscriber does not catch & retry UNAVAILABLE errors #4234
Description
A basic Pub/Sub message consumer stops consuming messages after a retryable error (see stack trace below, but in short _Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.UNAVAILABLE, The service was unable to fulfill your request. Please try again. [code=8a75])>). The app does not crash but the stream never recovers and continue to receive messages. Interesting observations;
- If I simply turn off WiFi on my laptop and run the same code, it keeps retrying until the machine is connected to the network and functions as expected. This tells me that this is a reaction to the specific StatusCode
- The exception sometimes happens on startup sometimes mid-stream.
Expected behavior:
- The application code would continue retrying to build the streamingPull connection and eventually recover and receive messages.
- This would be handled and surfaced as a warning, rather than a thread-killing exception.
This might be the same issue as 2683. This comment, in particular, seems like the solution that I would expect the client library to implement.
Answers to standard questions:
- OS type and version
MacOS Sierra 10.12.6 - Python version and virtual environment information
python --version
Python 2.7.10 (running in virtualenv) - google-cloud-python version
pip show google-cloud,pip show google-<service>orpip freeze
$ pip show google-cloud Name: google-cloud Version: 0.27.0 pip show google-cloud-pubsub Name: google-cloud-pubsub Version: 0.28.4
- Stacktrace if available
Exception in thread Consumer helper: consume bidirectional stream:
Traceback (most recent call last):
File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 810, in __bootstrap_inner
self.run()
File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 763, in run
self.__target(*self.__args, **self.__kwargs)
File "/Users/kir/cloud/env/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_consumer.py", line 248, in _blocking_consume
self._policy.on_exception(exc)
File "/Users/kir/cloud/env/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/policy/thread.py", line 140, in on_exception
raise exception
_Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.UNAVAILABLE, The service was unable to fulfill your request. Please try again. [code=8a75])>
- Steps to reproduce
- I was not able to reproduce this consistently. But it would happen ~1 in 10 times I ran the code.
- Code example
import time, datetime, sys
from google.cloud import pubsub_v1 as pubsub
subscription_name = "projects/%s/subscriptions/%s"%(sys.argv[1], sys.argv[2])
sleep_time_ms = 0
try:
sleep_time_ms = int(sys.argv[3])
except Exception:
print "Could not parse custom sleep time."
print "Using sleep time %g ms"%sleep_time_ms
def callback(message):
t = time.time()
time.sleep(float(sleep_time_ms)/1000)
print "Message " + message.data + " acked in %g second"%(time.time() - t)
message.ack()
subscription = pubsub.SubscriberClient().subscribe(subscription_name).open(callback=callback)
time.sleep(10000)
Activity
theacodes commentedon Oct 20, 2017
Rengerating pubsub to use api_core should solve this, I think.
kir-titievsky commentedon Oct 20, 2017
How would I do that?
theacodes commentedon Oct 20, 2017
We're working on it. :)
gcbirzan commentedon Oct 23, 2017
We're able to reproduce this consistently within 1-2 minutes of starting a snippet virtually identical to the one on https://cloud.google.com/pubsub/docs/pull#pubsub-pull-messages-async-python. There are no messages published on the topic. Our stacktrace is a bit different, though:
thomas-vl commentedon Oct 24, 2017
Any update about this issue?
anroopak commentedon Oct 24, 2017
Any update about this issue?
a13x6au3r commentedon Oct 24, 2017
Any update about this issue?
murataksoy commentedon Oct 25, 2017
@kir-titievsky, the solution you mentioned seems to work at first, but the problem with this approach seems to be that the CPU usage goes up every time the UNAVAILABLE exception is handled this way. Thus, a fix seems to be necessary.
lukesneeringer commentedon Oct 26, 2017
Hey folks,
I have an update on this issue. :-) I am working on it today.
As @murataksoy mentioned, it is not just as simple as adding an
ifline to return None fromUNAVAILABLE, because threads are building up. (#3965)Hoping to have a fix today or tomorrow.
sss721 commentedon Oct 27, 2017
Any update on this issue?
ckmvishnu commentedon Oct 31, 2017
Commented out exception raising code. But events are not consistent. I currently have subscriptions for insert/delete/preempt VM operations, But I consistently miss out preempt events.
My edit was to ignore self._policy.on_exception(exc) in _blocking_consume method.
lukesneeringer commentedon Oct 31, 2017
Sorry, I forgot to tag this issue. #4265 fixed this, and has been merged. A new version will be cut soon (within a day or two).
lukesneeringer commentedon Oct 31, 2017
Wait, sorry, I got confused. I still need to do the
api_coreintegration to properly fix this. Right now the exception is bubbling properly (that is what #4265 did) but not retrying.Re-opening, and a fix will be in soon.
43 remaining items
makrusak commentedon Nov 29, 2017
@dhermes My logs look like: https://pastebin.com/Dn9Lnynb (and so on and so on). As you can see initial request is sending repeatedly each 1.5 minutes.
Pyhon sample from here, for example: https://cloud.google.com/pubsub/docs/pull.
I am just using future like this instead of
dhermes commentedon Nov 30, 2017
@makrusak I just cut another release https://pypi.org/project/google-cloud-pubsub/0.29.2/. This will at least stop some of the threads that weren't previously being stopped. However, it won't close the bidirectional consumer(s) on failure. You can do that by calling
Policy.close()after the future returned fromPolicy.open()isdone().makrusak commentedon Nov 30, 2017
@dhermes Thanks! I'll validate it today and hope that all critical bugs will be fixed soon.
dhermes commentedon Nov 30, 2017
@makrusak I've definitely traced down at last one place where threads are leaked, but I can't get the CPU usage to spike much more than 3% after running for 300 seconds (which ends up recovering from
UNAVAILABLE3 times, leaving 3 zombie threads).itamar-resonai commentedon Dec 3, 2017
I've updated from 0.28.4 to 0.29.2, and now I get the following error exactly 90 seconds after the client starts listening:
dhermes commentedon Dec 4, 2017
@itamar-resonai That line is from
0.29.0, not from0.29.2.Maybe there is an issue with your install?
itamar-resonai commentedon Dec 4, 2017
Sorry, my bad.
It was indeed 0.29.2, but I didn't disable the
IgnoreRpcFailureHackpolicy-workaround, so that what happened...brianbaquiran commentedon May 29, 2018
It seems like the error still persists with pubsub
0.35.0. Mypip freezeI am seeing the traceback:
In case it matters, I'm running my code on GKE.
brianbaquiran commentedon May 30, 2018
Now, testing with
0.35.1, I am seeing the same error. Cutting-and-pasting from Stackdriver logs:My relevant
pip freeze:theacodes commentedon May 30, 2018
Thanks, @brianbaquiran, let's track this over at #4234.
jakeczyz commentedon Apr 6, 2019
@theacodes , the issue you linked to in the comment above is this self-same issue. :) Is there another issue you meant to link that discusses catching internal-thread pubsub exceptions? I haven't been able to find a satisfactory answer. Thanks.
https://stackoverflow.com/q/55552606/379037
PubSub: Making `thread.Policy.on_exception` more robust. (#4444)