apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-06 11:52:10 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Add rate limit exception catch to all consumers

Browse source
This commit is contained in:
Cyber MacGeddon 2024-08-19 22:12:01 +01:00
parent 387be1bbb4
commit 501e7131b7
2 changed files with 10 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

2
trustgraph/base/consumer.py
View file

@ -65,7 +65,7 @@ class Consumer(BaseProcessor):
self.consumer.negative_acknowledge(msg) self.consumer.negative_acknowledge(msg)
print("TooManyRequests: will retry") print("TooManyRequests: will retry")
__class__.processing_metric.labels(status="rate-limit").inc() __class__.processing_metric.labels(status="rate-limit").inc()
time.sleep(15) time.sleep(5)
continue continue
except Exception as e: except Exception as e:

9
trustgraph/base/consumer_producer.py
View file

@ -1,8 +1,10 @@
from pulsar.schema import JsonSchema from pulsar.schema import JsonSchema
from prometheus_client import Histogram, Info, Counter from prometheus_client import Histogram, Info, Counter
import time
from . base_processor import BaseProcessor from . base_processor import BaseProcessor
from .. exceptions import TooManyRequests
# FIXME: Derive from consumer? And producer? # FIXME: Derive from consumer? And producer?
@ -78,6 +80,13 @@ class ConsumerProducer(BaseProcessor):
__class__.processing_metric.labels(status="success").inc() __class__.processing_metric.labels(status="success").inc()
except TooManyRequests:
self.consumer.negative_acknowledge(msg)
print("TooManyRequests: will retry")
__class__.processing_metric.labels(status="rate-limit").inc()
time.sleep(5)
continue
except Exception as e: except Exception as e:
print("Exception:", e, flush=True) print("Exception:", e, flush=True)