Avoid return negative value in CounterMetric

server/src/main/java/org/elasticsearch/common/metrics/CounterMetric.java

@@ -8,29 +8,65 @@
 
 package org.elasticsearch.common.metrics;
 
+import org.elasticsearch.Assertions;
+
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 
-public class CounterMetric implements Metric {
 
+/**
+ * A {@link CounterMetric} is used to track the number of completed and outstanding items, for example, the number of executed refreshes,
+ * the currently used memory by indexing, the current pending search requests. In both cases, the current {@link CounterMetric#count} is
+ * always non-negative.
+ */
+public final class CounterMetric {
     private final LongAdder counter = new LongAdder();
+    private final AtomicLong assertingCounter = Assertions.ENABLED ? new AtomicLong() : null;
+
+    private boolean assertNonNegative(long n) {
+        assert n >= 0 : "CounterMetric value must always be non-negative; got: " + n;
+        return true;
+    }
 
     public void inc() {
         counter.increment();
+        assert assertNonNegative(assertingCounter.incrementAndGet());
     }
 
     public void inc(long n) {
         counter.add(n);
+        assert assertNonNegative(assertingCounter.addAndGet(n));
     }
 
     public void dec() {
         counter.decrement();
+        assert assertNonNegative(assertingCounter.decrementAndGet());
     }
 
     public void dec(long n) {
         counter.add(-n);
+        assert assertNonNegative(assertingCounter.addAndGet(-n));
     }
 
+    /**
+     * Returns the current count of this metric. The returned value is always non-negative.
+     * <p>
+     * As this metric is implemented using a {@link LongAdder}, the returned value is NOT an atomic snapshot;
+     * invocation in the absence of concurrent updates returns an accurate result, but concurrent updates that
+     * occur while the sum is being calculated might not be incorporated.
+     *
+     * @see LongAdder#sum()
+     */
     public long count() {
-        return counter.sum();
+        // The `counter.sum()` value is expected to always be non-negative. And if it's negative, then some concurrent updates
+        // aren't incorporated yet. In this case, we can immediately return 0L; but here we choose to retry several times
+        // to hopefully have a more accurate value than 0L.
+        for (int i = 0; i < 5; i++) {
+            final long count = counter.sum();
+            if (count >= 0L) {
+                return count;
+            }
+        }
+        return 0L;
     }
 }

server/src/main/java/org/elasticsearch/common/metrics/MeanMetric.java

@@ -10,7 +10,7 @@
 
 import java.util.concurrent.atomic.LongAdder;
 
-public class MeanMetric implements Metric {
+public final class MeanMetric {
 
     private final LongAdder counter = new LongAdder();
     private final LongAdder sum = new LongAdder();
@@ -20,13 +20,14 @@ public void inc(long n) {
         sum.add(n);
     }
 
-    public void dec(long n) {
-        counter.decrement();
-        sum.add(-n);
-    }
-
+    /**
+     * Returns the current count of this metric. This metric supports only {@link #inc(long)} that increases the counter
+     * whenever it's invoked; hence, the returned count is always non-negative.
+     */
     public long count() {
-        return counter.sum();
+        final long count = counter.sum();
+        assert count >= 0 : "Count of MeanMetric must always be non-negative; got " + count;
+        return count;
     }
 
     public long sum() {
@@ -40,9 +41,4 @@ public double mean() {
         }
         return 0.0;
     }
-
-    public void clear() {
-        counter.reset();
-        sum.reset();
-    }
 }

server/src/main/java/org/elasticsearch/index/search/stats/ShardSearchStats.java

@@ -71,10 +71,8 @@ public void onFailedQueryPhase(SearchContext searchContext) {
         computeStats(searchContext, statsHolder -> {
             if (searchContext.hasOnlySuggest()) {
                 statsHolder.suggestCurrent.dec();
-                assert statsHolder.suggestCurrent.count() >= 0;
             } else {
                 statsHolder.queryCurrent.dec();
-                assert statsHolder.queryCurrent.count() >= 0;
             }
         });
     }
@@ -85,11 +83,9 @@ public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
             if (searchContext.hasOnlySuggest()) {
                 statsHolder.suggestMetric.inc(tookInNanos);
                 statsHolder.suggestCurrent.dec();
-                assert statsHolder.suggestCurrent.count() >= 0;
             } else {
                 statsHolder.queryMetric.inc(tookInNanos);
                 statsHolder.queryCurrent.dec();
-                assert statsHolder.queryCurrent.count() >= 0;
             }
         });
     }
@@ -109,7 +105,6 @@ public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
         computeStats(searchContext, statsHolder -> {
             statsHolder.fetchMetric.inc(tookInNanos);
             statsHolder.fetchCurrent.dec();
-            assert statsHolder.fetchCurrent.count() >= 0;
         });
     }
 

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/slm/SnapshotLifecycleStatsTests.java

@@ -23,10 +23,10 @@ protected SnapshotLifecycleStats doParseInstance(XContentParser parser) throws I
 
     public static SnapshotLifecycleStats.SnapshotPolicyStats randomPolicyStats(String policyId) {
         return new SnapshotLifecycleStats.SnapshotPolicyStats(policyId,
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong());
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE));
     }
 
     public static SnapshotLifecycleStats randomLifecycleStats() {
@@ -37,10 +37,10 @@ public static SnapshotLifecycleStats randomLifecycleStats() {
             policyStats.put(policy, randomPolicyStats(policy));
         }
         return new SnapshotLifecycleStats(
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
             policyStats);
     }