Switch to AtomicLong for ingestCurrent metric to prevent negative values (#52581) (#52761)

Author: danhermann

server/src/main/java/org/elasticsearch/ingest/IngestMetric.java

@@ -22,6 +22,8 @@
 import org.elasticsearch.common.metrics.CounterMetric;
 import org.elasticsearch.common.metrics.MeanMetric;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 /**
  * <p>Metrics to measure ingest actions.
  * <p>This counts measure documents and timings for a given scope.
@@ -39,7 +41,7 @@ class IngestMetric {
      * The current count of things being measure. Should most likely ever be 0 or 1.
      * Useful when aggregating multiple metrics to see how many things are in flight.
      */
-    private final CounterMetric ingestCurrent = new CounterMetric();
+    private final AtomicLong ingestCurrent = new AtomicLong();
     /**
      * The ever increasing count of things being measured
      */
@@ -53,15 +55,15 @@ class IngestMetric {
      * Call this prior to the ingest action.
      */
     void preIngest() {
-        ingestCurrent.inc();
+        ingestCurrent.incrementAndGet();
     }
 
     /**
      * Call this after the performing the ingest action, even if the action failed.
      * @param ingestTimeInMillis The time it took to perform the action.
      */
     void postIngest(long ingestTimeInMillis) {
-        ingestCurrent.dec();
+        ingestCurrent.decrementAndGet();
         ingestTime.inc(ingestTimeInMillis);
         ingestCount.inc();
     }
@@ -90,6 +92,6 @@ void add(IngestMetric metrics) {
      * Creates a serializable representation for these metrics.
      */
     IngestStats.Stats createStats() {
-        return new IngestStats.Stats(ingestCount.count(), ingestTime.sum(), ingestCurrent.count(), ingestFailed.count());
+        return new IngestStats.Stats(ingestCount.count(), ingestTime.sum(), ingestCurrent.get(), ingestFailed.count());
     }
 }