ref: Rename backpressure metric (#303)

Rename `arroyo.consumer.paused.time` to `arroyo.consumer.backpressure.time`

Two main motivations for this:
- The new metric matches the name emitted by the Rust consumer, we should align these exactly to make the transition smoother
- The consumer is no longer actually paused immediately when backpressure happens (we now wait until we are in the backpressure state for at least a second as pausing is expensive). So the old metric name is kind of misleading now.

arroyo/processing/processor.py

@@ -66,7 +66,7 @@ class InvalidStateError(RuntimeError):
 ConsumerTiming = Literal[
     "arroyo.consumer.poll.time",
     "arroyo.consumer.processing.time",
-    "arroyo.consumer.paused.time",
+    "arroyo.consumer.backpressure.time",
     "arroyo.consumer.dlq.time",
     "arroyo.consumer.join.time",
     # This metric's timings overlap with DLQ/join time.
@@ -311,7 +311,7 @@ def run(self) -> None:
     def _clear_backpressure(self) -> None:
         if self.__backpressure_timestamp is not None:
             self.__metrics_buffer.incr_timing(
-                "arroyo.consumer.paused.time",
+                "arroyo.consumer.backpressure.time",
                 time.time() - self.__backpressure_timestamp,
             )
             self.__backpressure_timestamp = None

arroyo/utils/metric_defs.py

@@ -49,7 +49,7 @@
     # strategy.poll)
     "arroyo.consumer.processing.time",
     # Time (unitless) spent pausing the consumer due to backpressure (MessageRejected)
-    "arroyo.consumer.paused.time",
+    "arroyo.consumer.backpressure.time",
     # Time (unitless) spent in handling `InvalidMessage` exceptions and sending
     # messages to the the DLQ.
     "arroyo.consumer.dlq.time",

tests/processing/test_processor.py

@@ -136,7 +136,7 @@ def test_stream_processor_lifecycle() -> None:
         (Timing, "arroyo.consumer.processing.time"),
         (Increment, "arroyo.consumer.run.count"),
         (Timing, "arroyo.consumer.processing.time"),
-        (Timing, "arroyo.consumer.paused.time"),
+        (Timing, "arroyo.consumer.backpressure.time"),
         (Timing, "arroyo.consumer.join.time"),
         (Timing, "arroyo.consumer.shutdown.time"),
         (Timing, "arroyo.consumer.callback.time"),