Add test for cumulative async series reappearance; clarify DropAggregator epoch

- Test that when a cumulative async series stops reporting for a collection
  cycle and then reappears, its start time is anchored to the collection
  immediately preceding its reappearance (not instrument creation time).
- Add comment to DropAggregator.HANDLE explaining why creationEpochNanos=0
  is safe (the handle never produces data points).
Signed-off-by: Gregor Zeitlinger <gregor.zeitlinger@grafana.com>

Author: zeitlinger

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/DropAggregator.java

@@ -52,9 +52,11 @@ public List<? extends ExemplarData> getExemplars() {
 
   public static final Aggregator<PointData> INSTANCE = new DropAggregator();
 
+  // creationEpochNanos is 0 because DropAggregator never produces data points, so
+  // the start timestamp is irrelevant. A single shared HANDLE is safe to use.
   private static final AggregatorHandle<PointData> HANDLE =
       new AggregatorHandle<PointData>(
-          0, ExemplarReservoirFactory.noSamples(), /* isDoubleType= */ true) {
+          /* creationEpochNanos= */ 0, ExemplarReservoirFactory.noSamples(), /* isDoubleType= */ true) {
         @Override
         protected PointData doAggregateThenMaybeResetDoubles(
             long startEpochNanos,

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java

@@ -335,6 +335,73 @@ void collect_CumulativeReportsCumulativeObservations(MemoryMode memoryMode) {
                                 .hasAttributes(Attributes.builder().put("key", "value2").build())));
   }
 
+  @ParameterizedTest
+  @EnumSource(MemoryMode.class)
+  void collect_CumulativeSeriesDisappearsAndReappears(MemoryMode memoryMode) {
+    setup(memoryMode);
+
+    Attributes attrA = Attributes.empty();
+    Attributes attrB = Attributes.builder().put("key", "b").build();
+
+    // Collection 1: both series reported. start time = instrument creation time (START_SECOND_NANOS)
+    testClock.advance(Duration.ofSeconds(10));
+    longCounterStorage.record(attrA, 1);
+    longCounterStorage.record(attrB, 2);
+    assertThat(longCounterStorage.collect(resource, scope, testClock.now()))
+        .hasLongSumSatisfying(
+            sum ->
+                sum.isCumulative()
+                    .hasPointsSatisfying(
+                        point ->
+                            point
+                                .hasStartEpochNanos(START_SECOND_NANOS)
+                                .hasAttributes(attrA)
+                                .hasValue(1),
+                        point ->
+                            point
+                                .hasStartEpochNanos(START_SECOND_NANOS)
+                                .hasAttributes(attrB)
+                                .hasValue(2)));
+    registeredReader.setLastCollectEpochNanos(testClock.now());
+
+    // Collection 2: only attrA reported; attrB disappears and its handle is removed.
+    testClock.advance(Duration.ofSeconds(10));
+    longCounterStorage.record(attrA, 3);
+    assertThat(longCounterStorage.collect(resource, scope, testClock.now()))
+        .hasLongSumSatisfying(
+            sum ->
+                sum.isCumulative()
+                    .hasPointsSatisfying(
+                        point ->
+                            point
+                                .hasStartEpochNanos(START_SECOND_NANOS)
+                                .hasAttributes(attrA)
+                                .hasValue(3)));
+    registeredReader.setLastCollectEpochNanos(testClock.now());
+    long collectTime2 = testClock.now();
+
+    // Collection 3: attrB reappears. Its start time should be the time of the collection
+    // immediately preceding its reappearance (collectTime2), not the original creation time.
+    testClock.advance(Duration.ofSeconds(10));
+    longCounterStorage.record(attrA, 5);
+    longCounterStorage.record(attrB, 7);
+    assertThat(longCounterStorage.collect(resource, scope, testClock.now()))
+        .hasLongSumSatisfying(
+            sum ->
+                sum.isCumulative()
+                    .hasPointsSatisfying(
+                        point ->
+                            point
+                                .hasStartEpochNanos(START_SECOND_NANOS)
+                                .hasAttributes(attrA)
+                                .hasValue(5),
+                        point ->
+                            point
+                                .hasStartEpochNanos(collectTime2)
+                                .hasAttributes(attrB)
+                                .hasValue(7)));
+  }
+
   @ParameterizedTest
   @EnumSource(MemoryMode.class)
   void collect_DeltaComputesDiff(MemoryMode memoryMode) {