fix(uptime): Emit arroyo.consumer.latency when using thread-queue-parallel mode (#95919)

We need to emit this standard metric since we have our own custom offset
committing strategy. Without this, a bunch of datadog alerts fire.

<!-- Describe your PR here. -->

Author: wedamija

src/sentry/conf/types/kafka_definition.py

@@ -126,6 +126,9 @@ class ConsumerDefinition(TypedDict, total=False):
     # Hardcoded additional kwargs for strategy_factory
     static_args: Mapping[str, Any]
 
+    # Pass the consumer group ID to the strategy factory as 'consumer_group' kwarg
+    pass_consumer_group: bool
+
     require_synchronization: bool
     synchronize_commit_group_default: str
     synchronize_commit_log_topic_default: str

src/sentry/consumers/__init__.py

@@ -271,6 +271,7 @@ def ingest_transactions_options() -> list[click.Option]:
         "topic": Topic.UPTIME_RESULTS,
         "strategy_factory": "sentry.uptime.consumers.results_consumer.UptimeResultsStrategyFactory",
         "click_options": uptime_options(),
+        "pass_consumer_group": True,
     },
     "billing-metrics-consumer": {
         "topic": Topic.SNUBA_GENERIC_METRICS,
@@ -511,8 +512,14 @@ def get_stream_processor(
         name=consumer_name, params=list(consumer_definition.get("click_options") or ())
     )
     cmd_context = cmd.make_context(consumer_name, list(consumer_args))
+    extra_kwargs = {}
+    if consumer_definition.get("pass_consumer_group", False):
+        extra_kwargs["consumer_group"] = group_id
     strategy_factory = cmd_context.invoke(
-        strategy_factory_cls, **cmd_context.params, **consumer_definition.get("static_args") or {}
+        strategy_factory_cls,
+        **cmd_context.params,
+        **consumer_definition.get("static_args") or {},
+        **extra_kwargs,
     )
 
     def build_consumer_config(group_id: str):

src/sentry/remote_subscriptions/consumers/queue_consumer.py

@@ -7,6 +7,7 @@
 from collections import defaultdict
 from collections.abc import Callable
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any, Generic, TypeVar
 
 import sentry_sdk
@@ -30,25 +31,25 @@ class UnassignedPartitionError(Exception):
 
 @dataclass
 class WorkItem(Generic[T]):
-    """Work item that includes the original message for offset tracking."""
+    """Work item for processing with offset tracking information."""
 
     partition: Partition
     offset: int
+    timestamp: datetime
     result: T
-    message: Message[KafkaPayload | FilteredPayload]
 
 
 class OffsetTracker:
     """
     Tracks outstanding offsets and determines which offsets are safe to commit.
 
-    - Tracks offsets per partition
+    - Tracks offsets per partition with their timestamps
     - Only commits offsets when all prior offsets are processed
     - Thread-safe for concurrent access with per-partition locks
     """
 
     def __init__(self) -> None:
-        self.all_offsets: dict[Partition, set[int]] = defaultdict(set)
+        self.all_offsets: dict[Partition, dict[int, datetime]] = defaultdict(dict)
         self.outstanding: dict[Partition, set[int]] = defaultdict(set)
         self.last_committed: dict[Partition, int] = {}
         self.partition_locks: dict[Partition, threading.Lock] = {}
@@ -57,15 +58,15 @@ def _get_partition_lock(self, partition: Partition) -> threading.Lock:
         """Get the lock for a partition."""
         return self.partition_locks[partition]
 
-    def add_offset(self, partition: Partition, offset: int) -> None:
+    def add_offset(self, partition: Partition, offset: int, timestamp: datetime) -> None:
         """Record that we've started processing an offset."""
         if partition not in self.partition_locks:
             raise UnassignedPartitionError(
                 f"Partition {partition} is not assigned to this consumer"
             )
 
         with self._get_partition_lock(partition):
-            self.all_offsets[partition].add(offset)
+            self.all_offsets[partition][offset] = timestamp
             self.outstanding[partition].add(offset)
 
     def complete_offset(self, partition: Partition, offset: int) -> None:
@@ -76,11 +77,12 @@ def complete_offset(self, partition: Partition, offset: int) -> None:
         with self._get_partition_lock(partition):
             self.outstanding[partition].discard(offset)
 
-    def get_committable_offsets(self) -> dict[Partition, int]:
+    def get_committable_offsets(self) -> dict[Partition, tuple[int, datetime]]:
         """
         Get the highest offset per partition that can be safely committed.
 
-        For each partition, finds the highest contiguous offset that has been processed.
+        For each partition, finds the highest contiguous offset that has been processed,
+        and returns the timestamp of the oldest offset in the contiguous set.
         """
         committable = {}
         for partition in list(self.all_offsets.keys()):
@@ -92,20 +94,26 @@ def get_committable_offsets(self) -> dict[Partition, int]:
                 outstanding = self.outstanding[partition]
                 last_committed = self.last_committed.get(partition, -1)
 
-                min_offset = min(all_offsets)
-                max_offset = max(all_offsets)
+                offset_keys = set(all_offsets.keys())
+                min_offset = min(offset_keys)
+                max_offset = max(offset_keys)
 
                 start = max(last_committed + 1, min_offset)
 
                 highest_committable = last_committed
+                oldest_timestamp = None
+
                 for offset in range(start, max_offset + 1):
                     if offset in all_offsets and offset not in outstanding:
                         highest_committable = offset
+                        timestamp = all_offsets[offset]
+                        if oldest_timestamp is None or timestamp < oldest_timestamp:
+                            oldest_timestamp = timestamp
                     else:
                         break
 
-                if highest_committable > last_committed:
-                    committable[partition] = highest_committable
+                if highest_committable > last_committed and oldest_timestamp:
+                    committable[partition] = (highest_committable, oldest_timestamp)
 
         return committable
 
@@ -114,7 +122,9 @@ def mark_committed(self, partition: Partition, offset: int) -> None:
         with self._get_partition_lock(partition):
             self.last_committed[partition] = offset
             # Remove all offsets <= committed offset
-            self.all_offsets[partition] = {o for o in self.all_offsets[partition] if o > offset}
+            self.all_offsets[partition] = {
+                k: v for k, v in self.all_offsets[partition].items() if k > offset
+            }
 
     def clear(self) -> None:
         """Clear all offset tracking state."""
@@ -193,13 +203,15 @@ def __init__(
         self,
         result_processor: Callable[[str, T], None],
         identifier: str,
+        consumer_group: str,
         num_queues: int = 20,
         commit_interval: float = 1.0,
     ) -> None:
         self.result_processor = result_processor
         self.identifier = identifier
         self.num_queues = num_queues
         self.commit_interval = commit_interval
+        self.consumer_group = consumer_group
         self.offset_tracker = OffsetTracker()
         self.queues: list[queue.Queue[WorkItem[T]]] = []
         self.workers: list[OrderedQueueWorker[T]] = []
@@ -239,9 +251,21 @@ def _commit_loop(self) -> None:
                         len(committable),
                         tags={"identifier": self.identifier},
                     )
-
-                    self.commit_function(committable)
-                    for partition, offset in committable.items():
+                    for partition, (offset, oldest_timestamp) in committable.items():
+                        metrics.timing(
+                            "arroyo.consumer.latency",
+                            time.time() - oldest_timestamp.timestamp(),
+                            tags={
+                                "partition": partition.index,
+                                "kafka_topic": partition.topic.name,
+                                "consumer_group": self.consumer_group,
+                            },
+                        )
+
+                    self.commit_function(
+                        {partition: offset for partition, (offset, _) in committable.items()}
+                    )
+                    for partition, (offset, _) in committable.items():
                         self.offset_tracker.mark_committed(partition, offset)
             except Exception:
                 logger.exception("Error in commit loop")
@@ -257,7 +281,9 @@ def submit(self, group_key: str, work_item: WorkItem[T]) -> None:
         Submit a work item to the appropriate queue.
         """
         try:
-            self.offset_tracker.add_offset(work_item.partition, work_item.offset)
+            self.offset_tracker.add_offset(
+                work_item.partition, work_item.offset, work_item.timestamp
+            )
         except UnassignedPartitionError:
             logger.exception(
                 "Received message for unassigned partition, skipping",
@@ -400,10 +426,11 @@ def submit(self, message: Message[KafkaPayload | FilteredPayload]) -> None:
             assert isinstance(message.value, BrokerValue)
             partition = message.value.partition
             offset = message.value.offset
+            timestamp = message.value.timestamp
 
             if result is None:
                 try:
-                    self.queue_pool.offset_tracker.add_offset(partition, offset)
+                    self.queue_pool.offset_tracker.add_offset(partition, offset, timestamp)
                     self.queue_pool.offset_tracker.complete_offset(partition, offset)
                 except UnassignedPartitionError:
                     pass
@@ -414,21 +441,24 @@ def submit(self, message: Message[KafkaPayload | FilteredPayload]) -> None:
             work_item = WorkItem(
                 partition=partition,
                 offset=offset,
+                timestamp=timestamp,
                 result=result,
-                message=message,
             )
 
             self.queue_pool.submit(group_key, work_item)
 
         except Exception:
             logger.exception("Error submitting message to queue")
             if isinstance(message.value, BrokerValue):
-                self.queue_pool.offset_tracker.add_offset(
-                    message.value.partition, message.value.offset
-                )
-                self.queue_pool.offset_tracker.complete_offset(
-                    message.value.partition, message.value.offset
-                )
+                try:
+                    self.queue_pool.offset_tracker.add_offset(
+                        message.value.partition, message.value.offset, message.value.timestamp
+                    )
+                    self.queue_pool.offset_tracker.complete_offset(
+                        message.value.partition, message.value.offset
+                    )
+                except UnassignedPartitionError:
+                    pass
 
     def poll(self) -> None:
         stats = self.queue_pool.get_stats()

src/sentry/remote_subscriptions/consumers/result_consumer.py

@@ -105,6 +105,7 @@ class ResultsStrategyFactory(ProcessingStrategyFactory[KafkaPayload], Generic[T,
 
     def __init__(
         self,
+        consumer_group: str,
         mode: Literal["batched-parallel", "parallel", "serial", "thread-queue-parallel"] = "serial",
         max_batch_size: int | None = None,
         max_batch_time: int | None = None,
@@ -115,6 +116,7 @@ def __init__(
         commit_interval: float | None = None,
     ) -> None:
         self.mode = mode
+        self.consumer_group = consumer_group
         metric_tags = {"identifier": self.identifier, "mode": self.mode}
         self.result_processor = self.result_processor_cls()
         if mode == "batched-parallel":
@@ -134,6 +136,7 @@ def __init__(
             self.queue_pool = FixedQueuePool(
                 result_processor=self.result_processor,
                 identifier=self.identifier,
+                consumer_group=consumer_group,
                 num_queues=max_workers or 20,
                 commit_interval=commit_interval or 1.0,
             )