ref(consumers): Add min-partition-id

src/sentry/consumers/__init__.py

@@ -474,6 +474,7 @@ def get_stream_processor(
     group_instance_id: str | None = None,
     max_dlq_buffer_length: int | None = None,
     kafka_slice_id: int | None = None,
+    add_global_tags: bool = False,
 ) -> StreamProcessor:
     from sentry.utils import kafka_config
 
@@ -586,6 +587,9 @@ def build_consumer_config(group_id: str):
             stale_threshold_sec, strategy_factory
         )
 
+    if add_global_tags:
+        strategy_factory = MinPartitionMetricTagWrapper(strategy_factory)
+
     if healthcheck_file_path is not None:
         strategy_factory = HealthcheckStrategyFactoryWrapper(
             healthcheck_file_path, strategy_factory
@@ -641,6 +645,27 @@ def create_with_partitions(self, commit, partitions) -> ProcessingStrategy:
         return ValidateSchema(self.topic, self.enforce_schema, rv)
 
 
+class MinPartitionMetricTagWrapper(ProcessingStrategyFactory):
+    """
+    A wrapper that tracks the minimum partition index being processed by the consumer
+    and adds it as a global metric tag. This helps with monitoring partition distribution
+    and debugging partition-specific issues.
+    """
+
+    def __init__(self, inner: ProcessingStrategyFactory):
+        self.inner = inner
+
+    def create_with_partitions(self, commit, partitions):
+        from sentry.metrics.middleware import add_global_tags
+
+        # Update the min_partition global tag based on current partition assignment
+        if partitions:
+            min_partition = min(p.index for p in partitions)
+            add_global_tags(min_partition=str(min_partition))
+
+        return self.inner.create_with_partitions(commit, partitions)
+
+
 class HealthcheckStrategyFactoryWrapper(ProcessingStrategyFactory):
     def __init__(self, healthcheck_file_path: str, inner: ProcessingStrategyFactory):
         self.healthcheck_file_path = healthcheck_file_path

src/sentry/runner/commands/run.py

@@ -638,7 +638,12 @@ def basic_consumer(
         kafka_topic=topic, consumer_group=options["group_id"], kafka_slice_id=kafka_slice_id
     )
     processor = get_stream_processor(
-        consumer_name, consumer_args, topic=topic, kafka_slice_id=kafka_slice_id, **options
+        consumer_name,
+        consumer_args,
+        topic=topic,
+        kafka_slice_id=kafka_slice_id,
+        add_global_tags=True,
+        **options,
     )
 
     # for backwards compat: should eventually be removed