Add an assert to detect double decrementing of ingest metrics (#81450)

Author: joegallo

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

@@ -52,7 +52,8 @@ void preIngest() {
      * @param ingestTimeInNanos The time it took to perform the action.
      */
     void postIngest(long ingestTimeInNanos) {
-        ingestCurrent.decrementAndGet();
+        long current = ingestCurrent.decrementAndGet();
+        assert current >= 0 : "ingest metric current count double-decremented";
         this.ingestTimeInNanos.inc(ingestTimeInNanos);
         ingestCount.inc();
     }

server/src/test/java/org/elasticsearch/ingest/IngestMetricTests.java

@@ -31,4 +31,18 @@ public void testIngestTimeInNanos() {
         assertThat(1L, equalTo(metric.createStats().getIngestTimeInMillis()));
     }
 
+    public void testPostIngestDoubleDecrement() {
+        IngestMetric metric = new IngestMetric();
+
+        metric.preIngest();
+        assertThat(1L, equalTo(metric.createStats().getIngestCurrent()));
+
+        metric.postIngest(500000L);
+        assertThat(0L, equalTo(metric.createStats().getIngestCurrent()));
+
+        // the second postIngest triggers an assertion error
+        expectThrows(AssertionError.class, () -> metric.postIngest(500000L));
+        assertThat(-1L, equalTo(metric.createStats().getIngestCurrent()));
+    }
+
 }