feat: Increase backpressure threshold to 5 seconds

During inc-626 we saw excessive backpressure which caused
the consumer to be paused and resumed a lot. This led to
excessive network traffic as the local queue was purged and
messages were re-downloaded. Increase the threshold for pausing
to reduce the impact of this.

arroyo/processing/processor.py

@@ -34,6 +34,7 @@
 logger = logging.getLogger(__name__)
 
 METRICS_FREQUENCY_SEC = 1.0  # In seconds
+BACKPRESSURE_THRESHOLD = 5.0  # In seconds
 
 F = TypeVar("F", bound=Callable[[Any], Any])
 
@@ -148,7 +149,7 @@ def __init__(
 
         # The timestamp when backpressure state started
         self.__backpressure_timestamp: Optional[float] = None
-        # Consumer is paused after it is in backpressure state for > 1 second
+        # Consumer is paused after it is in backpressure state for > BACKPRESSURE_THRESHOLD seconds
         self.__is_paused = False
 
         self.__commit_policy_state = commit_policy.get_state_machine()
@@ -419,7 +420,7 @@ def _run_once(self) -> None:
                         self.__backpressure_timestamp = time.time()
 
                     elif not self.__is_paused and (
-                        time.time() - self.__backpressure_timestamp > 1
+                        time.time() - self.__backpressure_timestamp > BACKPRESSURE_THRESHOLD
                     ):
                         logger.debug(
                             "Caught %r while submitting %r, pausing consumer...",

rust-arroyo/src/processing/mod.rs

@@ -46,7 +46,7 @@ pub enum RunError {
     Strategy(#[source] Box<dyn std::error::Error>),
 }
 
-const BACKPRESSURE_THRESHOLD: Duration = Duration::from_secs(1);
+const BACKPRESSURE_THRESHOLD: Duration = Duration::from_secs(5);
 
 #[derive(Clone)]
 pub struct ConsumerState<TPayload>(Arc<(AtomicBool, Mutex<ConsumerStateInner<TPayload>>)>);
@@ -362,12 +362,12 @@ impl<TPayload: Clone + Send + Sync + 'static> StreamProcessor<TPayload> {
                     return Ok(());
                 };
 
-                // If we are in the backpressure state for more than 1 second,
+                // If we are in the backpressure state for more than BACKPRESSURE_THRESHOLD seconds,
                 // we pause the consumer and hold the message until it is
                 // accepted, at which point we can resume consuming.
                 if !consumer_is_paused && deadline.has_elapsed() {
                     tracing::warn!(
-                        "Consumer is in backpressure state for more than 1 second, pausing",
+                        "Consumer is in backpressure state for more than 5 seconds, pausing",
                     );
 
                     let partitions = self.consumer.tell().unwrap().into_keys().collect();