Add regression test for queue-full delta loss

Fills CoreMetricCollector's drain past its ArrayBlockingQueue capacity and
verifies every counter's delta is reported exactly once across cycles. Fails
against the pre-fix code (counters read-and-reset while the queue was full were
dropped) and passes with the remainingCapacity() guard.

Robust to the shared SpanMetricRegistryImpl singleton: tags its own
instrumentations with a unique prefix, counts only those, and drains
cycle-by-cycle until the queue is empty.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dougqh

internal-api/src/test/java/datadog/trace/api/telemetry/CoreMetricCollectorQueueFullTest.java

@@ -0,0 +1,91 @@
+package datadog.trace.api.telemetry;
+
+import static datadog.trace.api.telemetry.MetricCollector.RAW_QUEUE_SIZE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import datadog.trace.api.metrics.SpanMetricRegistryImpl;
+import datadog.trace.api.telemetry.CoreMetricCollector.CoreMetric;
+import java.util.Collection;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Regression guard for the queue-full delta-loss bug in {@link
+ * CoreMetricCollector#prepareMetrics()}.
+ *
+ * <p>Before the fix, the drain read a counter with {@code getValueAndReset()} — which clears the
+ * counter's delta baseline — and only then tried to {@code offer()} it onto a fixed-capacity queue.
+ * When the queue was already full the offer failed, so that already-reset delta was lost
+ * permanently rather than being carried to the next cycle. The fix checks {@code
+ * remainingCapacity()} before reading each counter and stops early, leaving the untouched counters
+ * for next time.
+ */
+class CoreMetricCollectorQueueFullTest {
+
+  /**
+   * Fills the drain past its queue capacity and verifies that every counter's delta is reported
+   * exactly once across as many cycles as it takes to empty — none dropped while the queue is full.
+   *
+   * <p>{@link CoreMetricCollector} is a singleton over the global {@link SpanMetricRegistryImpl},
+   * so this test can't assume a clean registry. It stays robust by tagging its own instrumentations
+   * with a unique prefix and counting only those, and it drains cycle-by-cycle until the queue
+   * yields nothing at all (so it never terminates early just because a cycle happened to surface
+   * only unrelated counters left behind by other tests). With the fix the sum of our own metrics
+   * equals the number we created; with the bug the ones skipped while the queue was full are lost,
+   * so the sum comes up short.
+   */
+  @Test
+  void skippedCountersRetainDeltaAcrossCycles() {
+    SpanMetricRegistryImpl registry = SpanMetricRegistryImpl.getInstance();
+    CoreMetricCollector collector = CoreMetricCollector.getInstance();
+
+    // Discard anything left in the queue by other tests sharing the singleton.
+    collector.drain();
+
+    // Unique instrumentation-name prefix so we count only the instrumentations this test creates,
+    // regardless of whatever else has already polluted the shared registry. CoreMetricCollector
+    // emits these as "integration_name:<name>" tags.
+    String namePrefix = "queue-full-test-" + System.nanoTime() + "-";
+    String tagPrefix = "integration_name:" + namePrefix;
+
+    // Create more non-zero counters than the queue can hold so at least one collection cycle
+    // overflows. A single onSpanCreated() per instrumentation leaves exactly one non-zero counter
+    // (spans_created) each; the rest read as 0 and are skipped.
+    int created = RAW_QUEUE_SIZE + 200;
+    for (int i = 0; i < created; i++) {
+      registry.get(namePrefix + i).onSpanCreated();
+    }
+
+    // Drain cycle-by-cycle until the queue is fully emptied, summing how many of our own metrics
+    // surface. Guard the loop so a bug that never converges fails loudly instead of hanging.
+    int seen = 0;
+    int cycles = 0;
+    while (true) {
+      collector.prepareMetrics();
+      Collection<CoreMetric> drained = collector.drain();
+      if (drained.isEmpty()) {
+        break;
+      }
+      for (CoreMetric metric : drained) {
+        if (isOurs(metric, tagPrefix)) {
+          seen++;
+        }
+      }
+      assertTrue(++cycles < 20, "drain did not converge; possible lost/looping delta");
+    }
+
+    assertEquals(
+        created,
+        seen,
+        "every counter's delta must be reported exactly once; a full queue must not drop any");
+  }
+
+  private static boolean isOurs(CoreMetric metric, String tagPrefix) {
+    for (String tag : metric.tags) {
+      if (tag.startsWith(tagPrefix)) {
+        return true;
+      }
+    }
+    return false;
+  }
+}