From 501e7131b771f0eae882ba22a17ef6acb2b3cf9b Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 19 Aug 2024 22:12:01 +0100 Subject: [PATCH] Add rate limit exception catch to all consumers --- trustgraph/base/consumer.py | 2 +- trustgraph/base/consumer_producer.py | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/trustgraph/base/consumer.py b/trustgraph/base/consumer.py index d5b9f133..0c975cf5 100644 --- a/trustgraph/base/consumer.py +++ b/trustgraph/base/consumer.py @@ -65,7 +65,7 @@ class Consumer(BaseProcessor): self.consumer.negative_acknowledge(msg) print("TooManyRequests: will retry") __class__.processing_metric.labels(status="rate-limit").inc() - time.sleep(15) + time.sleep(5) continue except Exception as e: diff --git a/trustgraph/base/consumer_producer.py b/trustgraph/base/consumer_producer.py index 06704d02..affa73b7 100644 --- a/trustgraph/base/consumer_producer.py +++ b/trustgraph/base/consumer_producer.py @@ -1,8 +1,10 @@ from pulsar.schema import JsonSchema from prometheus_client import Histogram, Info, Counter +import time from . base_processor import BaseProcessor +from .. exceptions import TooManyRequests # FIXME: Derive from consumer? And producer? @@ -78,6 +80,13 @@ class ConsumerProducer(BaseProcessor): __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: print("Exception:", e, flush=True)