fix(produce): Apply backpressure instead of crashing

If we get local queue full, let's raise MessageRejected to slow down the consumer.
getsentry · Aug 18, 2023 · 7a28935 · 7a28935
1 parent 6286c79
commit 7a28935
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/arroyo/processing/strategies/produce.py b/arroyo/processing/strategies/produce.py
@@ -87,7 +87,11 @@ def submit(
         future: Optional[Future[BrokerValue[TStrategyPayload]]] = None
 
         if not isinstance(message.payload, FilteredPayload):
-            future = self.__producer.produce(self.__topic, message.payload)
+            try:
+                future = self.__producer.produce(self.__topic, message.payload)
+            except BufferError as exc:
+                logger.exception(exc)
+                raise MessageRejected from exc
 
         self.__queue.append((message, future))