[da-vinci][server] Fix OTel duplicate attribute errors in KafkaConsumerServiceStats by adding region and pool type dimensions (linkedin#2695)

- Added two new OTel dimensions to all 12 KafkaConsumerServiceOtelMetricEntity metrics: VENICE_REGION_NAME and VENICE_CONSUMER_POOL_TYPE. This differentiates each instance's OTel attributes, matching Tehuti's per-region per-pool-type differentiation.
- Replaced the kafkaClusterAlias constructor parameter (a pre-concatenated regionAlias + poolType.getStatSuffix() string) with separate pubsubRegionAlias throughout the constructor chain.
-ConsumerPoolType implements VeniceDimensionInterface
- Added setRegionName() to OpenTelemetryMetricsSetup.Builder as a first-class builder method
- Added null validation to addCustomDimension(VeniceDimensionInterface) for consistency — throws IllegalArgumentException instead of implicit NPE.

Author: m-nagarajan

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/AggKafkaConsumerService.java

@@ -383,6 +383,11 @@ public synchronized AbstractKafkaConsumerService createKafkaConsumerService(fina
       return alreadyCreatedConsumerService;
     }
 
+    // getOrDefault returns the stored value (even if null) when the key IS present.
+    // kafkaClusterUrlToAliasMap can have null values when a kafka cluster config entry
+    // has a URL but no "name" field. Fall back to the resolved URL in that case.
+    String rawAlias = kafkaClusterUrlToAliasMap.getOrDefault(resolvedKafkaUrl, resolvedKafkaUrl);
+    String regionAlias = (rawAlias == null || rawAlias.isEmpty()) ? resolvedKafkaUrl : rawAlias;
     KafkaConsumerServiceDelegator.KafkaConsumerServiceBuilder consumerServiceBuilder =
         (poolSize, poolType) -> sharedConsumerAssignmentStrategy.constructor.construct(
             poolType,
@@ -392,7 +397,7 @@ public synchronized AbstractKafkaConsumerService createKafkaConsumerService(fina
             ingestionThrottler,
             kafkaClusterBasedRecordThrottler,
             metricsRepository,
-            kafkaClusterUrlToAliasMap.getOrDefault(resolvedKafkaUrl, resolvedKafkaUrl) + poolType.getStatSuffix(),
+            regionAlias,
             sharedConsumerNonExistingTopicCleanupDelayMS,
             staleTopicChecker,
             liveConfigBasedKafkaThrottlingEnabled,

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/ConsumerPoolType.java

@@ -1,6 +1,20 @@
 package com.linkedin.davinci.kafka.consumer;
 
-public enum ConsumerPoolType {
+import com.linkedin.venice.stats.dimensions.VeniceDimensionInterface;
+import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
+
+
+/**
+ * Consumer pool type for {@link KafkaConsumerService} instances.
+ *
+ * <p>Two string representations exist for each pool type:
+ * <ul>
+ *   <li>{@link #getStatSuffix()}: Tehuti stat name suffix (e.g., {@code _for_current_aa_wc_leader})</li>
+ *   <li>{@link #getDimensionValue()}: for OTel dimension value (e.g., {@code current_version_aa_wc_leader_pool})</li>
+ * </ul>
+ * These differ because Tehuti names are legacy and cannot change.
+ */
+public enum ConsumerPoolType implements VeniceDimensionInterface {
   REGULAR_POOL(""), // For other kinds of workload, and this pool type is also being used when using a single consumer
                     // pool.
 
@@ -22,4 +36,9 @@ public enum ConsumerPoolType {
   public String getStatSuffix() {
     return statSuffix;
   }
+
+  @Override
+  public VeniceMetricsDimensions getDimensionName() {
+    return VeniceMetricsDimensions.VENICE_CONSUMER_POOL_TYPE;
+  }
 }

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/KafkaConsumerService.java

@@ -115,7 +115,7 @@ protected KafkaConsumerService(
       final IngestionThrottler ingestionThrottler,
       final KafkaClusterBasedRecordThrottler kafkaClusterBasedRecordThrottler,
       final MetricsRepository metricsRepository,
-      final String kafkaClusterAlias,
+      final String pubsubRegionAlias,
       final long sharedConsumerNonExistingTopicCleanupDelayMS,
       final StaleTopicChecker staleTopicChecker,
       final boolean liveConfigBasedKafkaThrottlingEnabled,
@@ -152,7 +152,7 @@ protected KafkaConsumerService(
         ? statsOverride
         : createAggKafkaConsumerServiceStats(
             metricsRepository,
-            kafkaClusterAlias,
+            pubsubRegionAlias,
             this::getMaxElapsedTimeMSSinceLastPollInConsumerPool,
             metadataRepository,
             isUnregisterMetricForDeletedStoreEnabled,
@@ -437,19 +437,21 @@ public boolean hasAnySubscriptionFor(PubSubTopic versionTopic) {
 
   private AggKafkaConsumerServiceStats createAggKafkaConsumerServiceStats(
       MetricsRepository metricsRepository,
-      String kafkaClusterAlias,
+      String regionAlias,
       LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool,
       ReadOnlyStoreRepository metadataRepository,
       boolean isUnregisterMetricForDeletedStoreEnabled,
       String veniceClusterName) {
-    String nameWithKafkaClusterAlias = "kafka_consumer_service_for_" + kafkaClusterAlias;
+    String tehutiStatsNamePrefix = "kafka_consumer_service_for_" + regionAlias + poolType.getStatSuffix();
     return new AggKafkaConsumerServiceStats(
-        nameWithKafkaClusterAlias,
+        tehutiStatsNamePrefix,
         metricsRepository,
         metadataRepository,
         getMaxElapsedTimeSinceLastPollInConsumerPool,
         isUnregisterMetricForDeletedStoreEnabled,
-        veniceClusterName);
+        veniceClusterName,
+        regionAlias,
+        poolType);
   }
 
   @Override
@@ -565,7 +567,7 @@ KafkaConsumerService construct(
         IngestionThrottler ingestionThrottler,
         KafkaClusterBasedRecordThrottler kafkaClusterBasedRecordThrottler,
         MetricsRepository metricsRepository,
-        String kafkaClusterAlias,
+        String pubsubRegionAlias,
         long sharedConsumerNonExistingTopicCleanupDelayMS,
         StaleTopicChecker staleTopicChecker,
         boolean liveConfigBasedKafkaThrottlingEnabled,

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/PartitionWiseKafkaConsumerService.java

@@ -48,7 +48,7 @@ public class PartitionWiseKafkaConsumerService extends KafkaConsumerService {
       final IngestionThrottler ingestionThrottler,
       final KafkaClusterBasedRecordThrottler kafkaClusterBasedRecordThrottler,
       final MetricsRepository metricsRepository,
-      final String kafkaClusterAlias,
+      final String pubsubRegionAlias,
       final long sharedConsumerNonExistingTopicCleanupDelayMS,
       final StaleTopicChecker staleTopicChecker,
       final boolean liveConfigBasedKafkaThrottlingEnabled,
@@ -68,7 +68,7 @@ public class PartitionWiseKafkaConsumerService extends KafkaConsumerService {
         ingestionThrottler,
         kafkaClusterBasedRecordThrottler,
         metricsRepository,
-        kafkaClusterAlias,
+        pubsubRegionAlias,
         sharedConsumerNonExistingTopicCleanupDelayMS,
         staleTopicChecker,
         liveConfigBasedKafkaThrottlingEnabled,
@@ -91,7 +91,7 @@ public class PartitionWiseKafkaConsumerService extends KafkaConsumerService {
       final IngestionThrottler ingestionThrottler,
       final KafkaClusterBasedRecordThrottler kafkaClusterBasedRecordThrottler,
       final MetricsRepository metricsRepository,
-      final String kafkaClusterAlias,
+      final String pubsubRegionAlias,
       final long sharedConsumerNonExistingTopicCleanupDelayMS,
       final StaleTopicChecker staleTopicChecker,
       final boolean liveConfigBasedKafkaThrottlingEnabled,
@@ -112,7 +112,7 @@ public class PartitionWiseKafkaConsumerService extends KafkaConsumerService {
         ingestionThrottler,
         kafkaClusterBasedRecordThrottler,
         metricsRepository,
-        kafkaClusterAlias,
+        pubsubRegionAlias,
         sharedConsumerNonExistingTopicCleanupDelayMS,
         staleTopicChecker,
         liveConfigBasedKafkaThrottlingEnabled,

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreAwarePartitionWiseKafkaConsumerService.java

@@ -41,7 +41,7 @@ public class StoreAwarePartitionWiseKafkaConsumerService extends PartitionWiseKa
       final IngestionThrottler ingestionThrottler,
       final KafkaClusterBasedRecordThrottler kafkaClusterBasedRecordThrottler,
       final MetricsRepository metricsRepository,
-      final String kafkaClusterAlias,
+      final String pubsubRegionAlias,
       final long sharedConsumerNonExistingTopicCleanupDelayMS,
       final StaleTopicChecker staleTopicChecker,
       final boolean liveConfigBasedKafkaThrottlingEnabled,
@@ -61,7 +61,7 @@ public class StoreAwarePartitionWiseKafkaConsumerService extends PartitionWiseKa
         ingestionThrottler,
         kafkaClusterBasedRecordThrottler,
         metricsRepository,
-        kafkaClusterAlias,
+        pubsubRegionAlias,
         sharedConsumerNonExistingTopicCleanupDelayMS,
         staleTopicChecker,
         liveConfigBasedKafkaThrottlingEnabled,

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/AggKafkaConsumerServiceStats.java

@@ -1,5 +1,6 @@
 package com.linkedin.davinci.stats;
 
+import com.linkedin.davinci.kafka.consumer.ConsumerPoolType;
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.meta.ReadOnlyStoreRepository;
 import com.linkedin.venice.stats.AbstractVeniceAggStoreStats;
@@ -23,11 +24,17 @@ public AggKafkaConsumerServiceStats(
       ReadOnlyStoreRepository metadataRepository,
       LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool,
       boolean isUnregisterMetricForDeletedStoreEnabled,
-      String veniceClusterName) {
+      String veniceClusterName,
+      String pubsubRegionAlias,
+      ConsumerPoolType poolType) {
     super(
         regionName,
         metricsRepository,
-        new KafkaConsumerServiceStatsSupplier(getMaxElapsedTimeSinceLastPollInConsumerPool, veniceClusterName),
+        new KafkaConsumerServiceStatsSupplier(
+            getMaxElapsedTimeSinceLastPollInConsumerPool,
+            veniceClusterName,
+            pubsubRegionAlias,
+            poolType),
         metadataRepository,
         isUnregisterMetricForDeletedStoreEnabled,
         true);
@@ -97,12 +104,18 @@ public void recordTotalPartitionAssignmentForOtel(int partitionCount) {
   static class KafkaConsumerServiceStatsSupplier implements StatsSupplier<KafkaConsumerServiceStats> {
     private final LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool;
     private final String veniceClusterName;
+    private final String pubsubRegionAlias;
+    private final ConsumerPoolType poolType;
 
     KafkaConsumerServiceStatsSupplier(
         LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool,
-        String veniceClusterName) {
+        String veniceClusterName,
+        String pubsubRegionAlias,
+        ConsumerPoolType poolType) {
       this.getMaxElapsedTimeSinceLastPollInConsumerPool = getMaxElapsedTimeSinceLastPollInConsumerPool;
       this.veniceClusterName = veniceClusterName;
+      this.pubsubRegionAlias = pubsubRegionAlias;
+      this.poolType = poolType;
     }
 
     @Override
@@ -122,7 +135,9 @@ public KafkaConsumerServiceStats get(
           getMaxElapsedTimeSinceLastPollInConsumerPool,
           totalStats,
           SystemTime.INSTANCE,
-          veniceClusterName);
+          veniceClusterName,
+          pubsubRegionAlias,
+          poolType);
     }
   }
 }

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/KafkaConsumerServiceOtelMetricEntity.java

@@ -2,6 +2,8 @@
 
 import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CLUSTER_NAME;
 import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CONSUMER_POOL_ACTION;
+import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CONSUMER_POOL_TYPE;
+import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_REGION_NAME;
 import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME;
 import static com.linkedin.venice.utils.Utils.setOf;
 
@@ -21,48 +23,54 @@
 public enum KafkaConsumerServiceOtelMetricEntity implements ModuleMetricEntityInterface {
   POLL_BYTES(
       "ingestion.pubsub.consumer.poll.bytes", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.BYTES,
-      "Byte size of polled PubSub messages per poll request", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
+      "Byte size of polled PubSub messages per poll request",
+      setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   POLL_RECORD_COUNT(
       "ingestion.pubsub.consumer.poll.record_count", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.NUMBER,
-      "Number of records returned per poll request", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
+      "Number of records returned per poll request",
+      setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   POLL_COUNT(
       "ingestion.pubsub.consumer.poll.count", MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES, MetricUnit.NUMBER,
-      "Total count of poll requests to the PubSub consumer", setOf(VENICE_CLUSTER_NAME)
+      "Total count of poll requests to the PubSub consumer",
+      setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   POLL_TIME(
       "ingestion.pubsub.consumer.poll.time", MetricType.HISTOGRAM, MetricUnit.MILLISECOND,
-      "Latency of PubSub consumer poll requests", setOf(VENICE_CLUSTER_NAME)
+      "Latency of PubSub consumer poll requests",
+      setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   POLL_NON_EMPTY_COUNT(
       "ingestion.pubsub.consumer.poll.non_empty_count", MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES, MetricUnit.NUMBER,
-      "Count of poll requests that returned at least one record", setOf(VENICE_CLUSTER_NAME)
+      "Count of poll requests that returned at least one record",
+      setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   POLL_ERROR_COUNT(
       "ingestion.pubsub.consumer.poll.error_count", MetricType.COUNTER, MetricUnit.NUMBER,
-      "Count of PubSub consumer poll errors", setOf(VENICE_CLUSTER_NAME)
+      "Count of PubSub consumer poll errors", setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   PRODUCE_TO_WRITE_BUFFER_TIME(
       "ingestion.pubsub.consumer.produce_to_write_buffer_time", MetricType.HISTOGRAM, MetricUnit.MILLISECOND,
-      "Latency of producing consumed records to the write buffer", setOf(VENICE_CLUSTER_NAME)
+      "Latency of producing consumed records to the write buffer",
+      setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   TOPIC_DETECTED_DELETED_COUNT(
       "ingestion.pubsub.consumer.topic.detected_deleted_count", MetricType.COUNTER, MetricUnit.NUMBER,
-      "Count of topics detected as deleted", setOf(VENICE_CLUSTER_NAME)
+      "Count of topics detected as deleted", setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   ORPHAN_TOPIC_PARTITION_COUNT(
       "ingestion.pubsub.consumer.orphan_subscription_count", MetricType.COUNTER, MetricUnit.NUMBER,
       "Count of orphan subscriptions, topic partitions assigned to a consumer with no running ingestion task",
-      setOf(VENICE_CLUSTER_NAME)
+      setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   /**
@@ -72,7 +80,7 @@ public enum KafkaConsumerServiceOtelMetricEntity implements ModuleMetricEntityIn
   POOL_ACTION_TIME(
       "ingestion.pubsub.consumer.pool_action.time", MetricType.HISTOGRAM, MetricUnit.MILLISECOND,
       "Latency of consumer pool actions (subscribe, update assignment)",
-      setOf(VENICE_CLUSTER_NAME, VENICE_CONSUMER_POOL_ACTION)
+      setOf(VENICE_CLUSTER_NAME, VENICE_CONSUMER_POOL_ACTION, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   /**
@@ -86,14 +94,14 @@ public enum KafkaConsumerServiceOtelMetricEntity implements ModuleMetricEntityIn
   POLL_TIME_SINCE_LAST_SUCCESS(
       "ingestion.pubsub.consumer.poll.time_since_last_success", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS,
       MetricUnit.MILLISECOND, "Max elapsed time since last successful poll across consumers in the pool",
-      setOf(VENICE_CLUSTER_NAME)
+      setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   ),
 
   /** Raw per-consumer partition assignment counts. OTel-only (Tehuti uses 4 pre-computed gauges). */
   PARTITION_ASSIGNMENT_COUNT(
       "ingestion.pubsub.consumer.partition_assignment.count", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS,
       MetricUnit.NUMBER, "Raw per-consumer partition assignment counts across the consumer pool",
-      setOf(VENICE_CLUSTER_NAME)
+      setOf(VENICE_CLUSTER_NAME, VENICE_REGION_NAME, VENICE_CONSUMER_POOL_TYPE)
   );
 
   private final MetricEntity metricEntity;