atlas: fix expiration for counter batch updates (#1178)

Ensure that if a counter is expired, any existing batch
updaters will re-lookup the counter when flushing updates.
Otherwise all updates from the batch updater will get
ignored.

Author: brharrington

spectator-reg-atlas/src/main/java/com/netflix/spectator/atlas/AtlasCounter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2024 Netflix, Inc.
+ * Copyright 2014-2025 Netflix, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -56,4 +56,10 @@ class AtlasCounter extends AtlasMeter implements Counter {
   @Override public double actualCount() {
     return value.poll();
   }
+
+  @Override public Counter.BatchUpdater batchUpdater(int batchSize) {
+    AtlasCounterBatchUpdater updater = new AtlasCounterBatchUpdater(batchSize);
+    updater.accept(() -> this);
+    return updater;
+  }
 }

spectator-reg-atlas/src/main/java/com/netflix/spectator/atlas/AtlasCounterBatchUpdater.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2014-2025 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spectator.atlas;
+
+import com.netflix.spectator.api.Counter;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+final class AtlasCounterBatchUpdater implements Counter.BatchUpdater, Consumer<Supplier<Counter>> {
+
+  private Supplier<Counter> counterSupplier;
+  private final int batchSize;
+
+  private int count;
+  private double sum;
+
+  AtlasCounterBatchUpdater(int batchSize) {
+    this.batchSize = batchSize;
+    this.count = 0;
+    this.sum = 0.0;
+  }
+
+  @Override
+  public void accept(Supplier<Counter> counterSupplier) {
+    this.counterSupplier = counterSupplier;
+  }
+
+  private AtlasCounter getCounter() {
+    if (counterSupplier != null) {
+      Counter c = counterSupplier.get();
+      return (c instanceof AtlasCounter) ? (AtlasCounter) c : null;
+    }
+    return null;
+  }
+
+  @Override
+  public void add(double amount) {
+    if (Double.isFinite(amount) && amount > 0.0) {
+      sum += amount;
+      ++count;
+      if (count >= batchSize) {
+        flush();
+      }
+    }
+  }
+
+  @Override
+  public void flush() {
+    AtlasCounter counter = getCounter();
+    if (counter != null) {
+      counter.add(sum);
+      sum = 0.0;
+      count = 0;
+    }
+  }
+
+  @Override
+  public void close() throws Exception {
+    flush();
+  }
+}

spectator-reg-atlas/src/test/java/com/netflix/spectator/atlas/AtlasCounterTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 Netflix, Inc.
+ * Copyright 2014-2025 Netflix, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,6 +15,7 @@
  */
 package com.netflix.spectator.atlas;
 
+import com.netflix.spectator.api.Counter;
 import com.netflix.spectator.api.Id;
 import com.netflix.spectator.api.ManualClock;
 import com.netflix.spectator.api.Measurement;
@@ -124,4 +125,24 @@ public void preferStatisticFromTags() {
     Id actual = c.measure().iterator().next().id();
     Assertions.assertEquals(id.withTag(DsType.rate), actual);
   }
+
+  @Test
+  public void batchUpdate() throws Exception {
+    try (Counter.BatchUpdater b = counter.batchUpdater(2)) {
+      b.increment();
+      b.add(Double.POSITIVE_INFINITY);
+      b.add(Double.NaN);
+      b.add(-1.0);
+      clock.setWallTime(step + 1);
+      Assertions.assertEquals(0, counter.count());
+      b.increment();
+      clock.setWallTime(step * 2 + 1);
+      Assertions.assertEquals(2, counter.count());
+      b.increment(42);
+      clock.setWallTime(step * 3 + 1);
+      Assertions.assertEquals(0, counter.count());
+    }
+    clock.setWallTime(step * 4 + 1);
+    Assertions.assertEquals(42, counter.count());
+  }
 }

spectator-reg-atlas/src/test/java/com/netflix/spectator/atlas/AtlasRegistryTest.java

@@ -20,6 +20,7 @@
 import com.fasterxml.jackson.core.JsonToken;
 import com.fasterxml.jackson.dataformat.smile.SmileFactory;
 import com.netflix.spectator.api.Clock;
+import com.netflix.spectator.api.Counter;
 import com.netflix.spectator.api.Gauge;
 import com.netflix.spectator.api.Id;
 import com.netflix.spectator.api.ManualClock;
@@ -285,6 +286,19 @@ public void flushOnShutdownCounter() {
     Assertions.assertEquals(1.0, getValue(payloads.get(1)));
   }
 
+  @Test
+  public void batchUpdateExpiration() throws Exception {
+    Counter.BatchUpdater b = registry.counter("test").batchUpdater(2);
+    final long step = 10_000L;
+    final long ttl = step * 6 * 20;
+    clock.setWallTime(step * ttl);
+    registry.removeExpiredMeters();
+    b.increment();
+    b.flush();
+    clock.setWallTime(step * (ttl + 1));
+    Assertions.assertEquals(1, registry.counter("test").count());
+  }
+
   private double getValue(PublishPayload payload) {
     return payload.getMetrics()
         .stream()