[server][da-vinci] Fix negative OTel counters by using cumulative sum() in

async counter callbacks

The ObservableLongCounter callback in MetricEntityState.reportToMeasurement()
used LongAdder.sumThenReset() which reports deltas. The OTel SDK expects
cumulative values from async counter callbacks and computes its own deltas
internally (current - previous). This caused delta-of-delta, producing
negative counter values when traffic varied between collection intervals.

Fix: replace sumThenReset() with sum() so the callback reports cumulative
values. Remove sumThenReset() from MetricAttributesData entirely. Add
negative-value guard for ASYNC_COUNTER_FOR_HIGH_PERF_CASES (silently dropped,
matching OTel SDK behavior for synchronous LongCounter).

Add reusable multi-collection test utilities in OpenTelemetryDataTestUtils
for both ASYNC_COUNTER and ASYNC_UP_DOWN_COUNTER, validating DELTA and
CUMULATIVE temporality. Add regression tests across DIVStats,
KafkaConsumerServiceStats, IngestionOtelStats, and ServerReadQuotaUsageStats.

Author: m-nagarajan

clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/DIVStatsOtelTest.java

@@ -214,7 +214,7 @@ public void testDIVResultDimensionIsolation() {
     stats.recordSuccessMsg(TEST_STORE_NAME, 1);
     stats.recordDuplicateMsg(TEST_STORE_NAME, 1);
 
-    // Collect once — ASYNC_COUNTER uses sumThenReset, so subsequent collections would see 0.
+    // Collect once to snapshot all dimension values from a single collection interval.
     Collection<MetricData> metrics = inMemoryMetricReader.collectAllMetrics();
     validateAsyncCounterFromCollection(
         metrics,
@@ -549,9 +549,7 @@ private void validateAsyncCounter(String metricName, long expectedValue, Attribu
 
   /**
    * Validates an async counter value from a pre-collected metrics snapshot. Use this when
-   * validating multiple dimension values of the same ASYNC_COUNTER metric in a single test —
-   * each {@code collectAllMetrics()} call resets the LongAdder via {@code sumThenReset()},
-   * so only the first collection returns the accumulated values.
+   * validating multiple dimension values of the same ASYNC_COUNTER metric in a single test.
    */
   private static void validateAsyncCounterFromCollection(
       Collection<MetricData> metricsData,
@@ -564,4 +562,41 @@ private static void validateAsyncCounterFromCollection(
     assertEquals(point.getValue(), expectedValue);
   }
 
+  /**
+   * Verifies that DIV MESSAGE_COUNT (ASYNC_COUNTER_FOR_HIGH_PERF_CASES) produces correct data
+   * across multiple collection intervals under both DELTA and CUMULATIVE temporality.
+   */
+  @Test
+  public void testMessageCountMultiCollection() {
+    Attributes attrs = buildMessageAttributes(TEST_STORE_NAME, VersionRole.CURRENT, VeniceDIVResult.SUCCESS);
+    OpenTelemetryDataTestUtils.validateAsyncCounterMultiCollection(
+        TEST_METRIC_PREFIX,
+        SERVER_METRIC_ENTITIES,
+        OTEL_MESSAGE_COUNT,
+        attrs,
+        repo -> {
+          AggVersionedDIVStats s = createDIVStatsWithStore(repo, TEST_STORE_NAME);
+          return n -> {
+            for (int i = 0; i < n; i++)
+              s.recordSuccessMsg(TEST_STORE_NAME, 1);
+          };
+        },
+        new long[] { 5, 2, 8 });
+  }
+
+  /** Creates an AggVersionedDIVStats with a single store set up (version 1=CURRENT). */
+  private AggVersionedDIVStats createDIVStatsWithStore(VeniceMetricsRepository repo, String storeName) {
+    ReadOnlyStoreRepository mockRepo = mock(ReadOnlyStoreRepository.class);
+    doReturn(Collections.emptyList()).when(mockRepo).getAllStores();
+    AggVersionedDIVStats divStats = new AggVersionedDIVStats(repo, mockRepo, true, TEST_CLUSTER_NAME);
+    Store mockStore = mock(Store.class);
+    doReturn(storeName).when(mockStore).getName();
+    doReturn(1).when(mockStore).getCurrentVersion();
+    Version v1 = new VersionImpl(storeName, 1, "push-1");
+    v1.setStatus(VersionStatus.ONLINE);
+    doReturn(Collections.singletonList(v1)).when(mockStore).getVersions();
+    divStats.handleStoreCreated(mockStore);
+    return divStats;
+  }
+
 }

clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/KafkaConsumerServiceStatsOtelTest.java

@@ -769,4 +769,33 @@ private double getTehutiMetricValue(String statsName, String sensorName, String
     assertNotNull(metric, "Tehuti metric should exist: " + metricName);
     return metric.value();
   }
+
+  /**
+   * Verifies that POLL_COUNT (ASYNC_COUNTER_FOR_HIGH_PERF_CASES) produces correct data
+   * across multiple collection intervals under both DELTA and CUMULATIVE temporality.
+   */
+  @Test
+  public void testPollCountMultiCollection() {
+    OpenTelemetryDataTestUtils.validateAsyncCounterMultiCollection(
+        TEST_METRIC_PREFIX,
+        SERVER_METRIC_ENTITIES,
+        POLL_COUNT.getMetricName(),
+        buildNonStoreAttributes(),
+        repo -> {
+          KafkaConsumerServiceStats s = new KafkaConsumerServiceStats(
+              repo,
+              TOTAL_STORE_NAME,
+              () -> 42L,
+              null,
+              SystemTime.INSTANCE,
+              TEST_CLUSTER_NAME,
+              TEST_REGION_NAME,
+              TEST_POOL_TYPE);
+          return n -> {
+            for (int i = 0; i < n; i++)
+              s.recordPollRequestLatency(5.0);
+          };
+        },
+        new long[] { 10, 4, 15 });
+  }
 }

clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStatsTest.java

@@ -88,6 +88,7 @@
 import com.linkedin.venice.stats.dimensions.VenicePartialUpdateOperation;
 import com.linkedin.venice.stats.dimensions.VeniceRecordType;
 import com.linkedin.venice.stats.dimensions.VeniceRegionLocality;
+import com.linkedin.venice.utils.OpenTelemetryDataTestUtils;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
 import io.tehuti.metrics.MetricsRepository;
@@ -1367,4 +1368,23 @@ private Map<Integer, Integer> getPushTimeoutByVersion(IngestionOtelStats stats)
   private interface ReflectiveResolveBackup {
     int resolve() throws Exception;
   }
+
+  /**
+   * Verifies that ingestion ASYNC_COUNTER_FOR_HIGH_PERF_CASES metrics produce correct data
+   * across multiple collection intervals under both DELTA and CUMULATIVE temporality.
+   */
+  @Test
+  public void testRecordsConsumedMultiCollection() {
+    OpenTelemetryDataTestUtils.validateAsyncCounterMultiCollection(
+        TEST_PREFIX,
+        SERVER_METRIC_ENTITIES,
+        INGESTION_RECORDS_CONSUMED.getMetricEntity().getMetricName(),
+        buildAttributesWithVersionRoleAndReplicaType(VersionRole.CURRENT, ReplicaType.LEADER),
+        repo -> {
+          IngestionOtelStats s = createStats(repo);
+          s.updateVersionInfo(CURRENT_VERSION, FUTURE_VERSION);
+          return n -> s.recordRecordsConsumed(CURRENT_VERSION, ReplicaType.LEADER, (int) n);
+        },
+        new long[] { 500, 100, 800 });
+  }
 }

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/metrics/MetricAttributesData.java

@@ -15,7 +15,7 @@
  *
  * <p>The {@link LongAdder} provides high-throughput recording capability by minimizing contention
  * across threads. The accumulated value is read during OpenTelemetry's metric collection callback
- * via {@link #sumThenReset()}.
+ * via {@link #sum()} (for ObservableLongCounter/UpDownCounter callbacks that must report cumulative values).
  */
 public class MetricAttributesData {
   private final Attributes attributes;
@@ -67,13 +67,20 @@ public void add(long value) {
   }
 
   /**
-   * Returns the current sum and resets the adder to zero.
-   * This is typically called during OpenTelemetry's metric collection callback.
-   * Only call this for observable counter metrics where adder is guaranteed non-null.
+   * Returns the current cumulative sum without resetting.
+   * This is the correct method to use in OpenTelemetry's ObservableLongCounter/UpDownCounter callbacks,
+   * which must report <b>cumulative</b> values per the OTel spec. The SDK handles delta computation
+   * internally based on the configured aggregation temporality.
+   *
+   * <p>Using {@code sumThenReset()} in observable counter callbacks causes the SDK to compute
+   * delta-of-delta (because it subtracts the previous observation from the current one, but
+   * {@code sumThenReset()} already returns a delta), producing negative counter values when
+   * traffic varies between collection intervals.
    *
-   * @return the sum before reset
+   * @return the current cumulative sum
    */
-  public long sumThenReset() {
-    return adder.sumThenReset();
+  public long sum() {
+    return adder.sum();
   }
+
 }

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/metrics/MetricEntityState.java

@@ -27,6 +27,7 @@
  */
 public abstract class MetricEntityState extends AsyncMetricEntityState {
   private final boolean isObservableCounter;
+  private final boolean isMonotonicCounter;
   /** define both long and double consumer to avoid unnecessary conversions **/
   private final ObjDoubleConsumer<MetricAttributesData> otelDoubleRecordingStrategy;
   private final ObjLongConsumer<MetricAttributesData> otelLongRecordingStrategy;
@@ -49,6 +50,7 @@ public MetricEntityState(
         null);
     MetricType metricType = metricEntity.getMetricType();
     this.isObservableCounter = metricType.isObservableCounterType();
+    this.isMonotonicCounter = metricType == MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES;
     this.otelDoubleRecordingStrategy = createOtelDoubleRecordingStrategy(metricType);
     this.otelLongRecordingStrategy = createOtelLongRecordingStrategy(metricType);
   }
@@ -86,6 +88,12 @@ protected final void registerObservableCounterIfNeeded() {
   /**
    * Reports all accumulated values to the OpenTelemetry measurement.
    * This is the callback invoked by OTel during metric collection.
+   *
+   * <p>Uses {@link MetricAttributesData#sum()} (not {@code sumThenReset()}) because the OTel spec
+   * requires ObservableLongCounter/UpDownCounter callbacks to report <b>cumulative</b> values. The
+   * SDK handles delta computation internally based on the configured aggregation temporality.
+   * Using {@code sumThenReset()} caused the SDK to compute delta-of-delta, producing negative
+   * counter values when traffic varied between collection intervals.
    */
   private void reportToMeasurement(ObservableLongMeasurement measurement) {
     Iterable<MetricAttributesData> allData = getAllMetricAttributesData();
@@ -95,12 +103,15 @@ private void reportToMeasurement(ObservableLongMeasurement measurement) {
 
     for (MetricAttributesData holder: allData) {
       if (holder.hasAdder()) {
-        long value = holder.sumThenReset();
-        // Skip zero values to avoid polluting metrics with stale attribute combinations
-        // (e.g., from deleted stores) rather than trying to clean up all the registered
-        // callbacks which could be complex. For delta-temporality async counters, omitting
-        // a zero report correctly means "no change in this period."
-        if (value != 0) {
+        long value = holder.sum();
+        // For monotonic counters (ASYNC_COUNTER), skip attribute combinations that were never
+        // recorded (cumulative sum == 0, since values are always non-negative). For stale
+        // combinations (e.g., deleted stores), the cumulative sum stays constant and the SDK
+        // correctly computes delta=0.
+        // For up-down counters (ASYNC_UP_DOWN_COUNTER), always report — a cumulative sum of 0
+        // is a legitimate value (e.g., all connections opened have been closed) and must be
+        // observed by the SDK to compute the correct delta.
+        if (value != 0 || !isMonotonicCounter) {
           measurement.record(value, holder.getAttributes());
         }
       }
@@ -132,6 +143,11 @@ private ObjDoubleConsumer<MetricAttributesData> createOtelDoubleRecordingStrateg
   private ObjLongConsumer<MetricAttributesData> createOtelLongRecordingStrategy(MetricType metricType) {
     switch (metricType) {
       case ASYNC_COUNTER_FOR_HIGH_PERF_CASES:
+        return (holder, value) -> {
+          if (value >= 0) {
+            holder.add(value);
+          }
+        };
       case ASYNC_UP_DOWN_COUNTER_FOR_HIGH_PERF_CASES:
         return (holder, value) -> holder.add(value);
       case COUNTER: