aggregator reset tests

Author: fandreuz

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

@@ -502,4 +502,25 @@ void enabledThenDisableThenEnable_recordAndCollect(MemoryMode memoryMode) {
 
     assertThat(longCounterStorage.collect(resource, scope, 0, 0).isEmpty()).isFalse();
   }
+
+  @ParameterizedTest
+  @EnumSource(MemoryMode.class)
+  void disableDropsAggregatorState(MemoryMode memoryMode) {
+    setup(memoryMode);
+
+    longCounterStorage.record(Attributes.empty(), 10);
+
+    longCounterStorage.setEnabled(false);
+    longCounterStorage.setEnabled(true);
+
+    longCounterStorage.record(Attributes.empty(), 5);
+
+    MetricData metricData = longCounterStorage.collect(resource, scope, 0, 0);
+    assertThat(metricData)
+        .hasLongSumSatisfying(
+            sum ->
+                sum.satisfies(
+                    sumData ->
+                        assertThat(sumData.getPoints()).allMatch(point -> point.getValue() == 5d)));
+  }
 }

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

@@ -897,4 +897,29 @@ void enabledThenDisableThenEnable_recordAndCollect(MemoryMode memoryMode) {
 
     assertThat(storage.collect(RESOURCE, INSTRUMENTATION_SCOPE_INFO, 0, 10).isEmpty()).isFalse();
   }
+
+  @ParameterizedTest
+  @EnumSource(MemoryMode.class)
+  void disableDropsAggregatorState(MemoryMode memoryMode) {
+    initialize(memoryMode);
+
+    DefaultSynchronousMetricStorage<?, ?> storage =
+        new DefaultSynchronousMetricStorage<>(
+            deltaReader, METRIC_DESCRIPTOR, aggregator, attributesProcessor, CARDINALITY_LIMIT);
+
+    storage.recordDouble(10d, Attributes.empty(), Context.current());
+
+    storage.setEnabled(false);
+    storage.setEnabled(true);
+
+    storage.recordDouble(5d, Attributes.empty(), Context.current());
+
+    MetricData metricData = storage.collect(RESOURCE, INSTRUMENTATION_SCOPE_INFO, 0, 10);
+    assertThat(metricData)
+        .hasDoubleSumSatisfying(
+            sum ->
+                sum.satisfies(
+                    sumData ->
+                        assertThat(sumData.getPoints()).allMatch(point -> point.getValue() == 5d)));
+  }
 }