KAFKA-8696: clean up Sum/Count/Total metrics

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java

@@ -42,9 +42,9 @@
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.Avg;
-import org.apache.kafka.common.metrics.stats.Count;
 import org.apache.kafka.common.metrics.stats.Max;
 import org.apache.kafka.common.metrics.stats.Meter;
+import org.apache.kafka.common.metrics.stats.WindowedCount;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.FindCoordinatorRequest;
 import org.apache.kafka.common.requests.FindCoordinatorRequest.CoordinatorType;
@@ -961,7 +961,7 @@ public void onSuccess(ClientResponse clientResponse, RequestFuture<T> future) {
     }
 
     protected Meter createMeter(Metrics metrics, String groupName, String baseName, String descriptiveName) {
-        return new Meter(new Count(),
+        return new Meter(new WindowedCount(),
                 metrics.metricName(baseName + "-rate", groupName,
                         String.format("The number of %s per second", descriptiveName)),
                 metrics.metricName(baseName + "-total", groupName,

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

@@ -47,10 +47,10 @@
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.Avg;
-import org.apache.kafka.common.metrics.stats.Count;
 import org.apache.kafka.common.metrics.stats.Max;
 import org.apache.kafka.common.metrics.stats.Meter;
 import org.apache.kafka.common.metrics.stats.Min;
+import org.apache.kafka.common.metrics.stats.WindowedCount;
 import org.apache.kafka.common.metrics.stats.Value;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
@@ -1657,7 +1657,7 @@ private FetchManagerMetrics(Metrics metrics, FetcherMetricsRegistry metricsRegis
             this.fetchLatency = metrics.sensor("fetch-latency");
             this.fetchLatency.add(metrics.metricInstance(metricsRegistry.fetchLatencyAvg), new Avg());
             this.fetchLatency.add(metrics.metricInstance(metricsRegistry.fetchLatencyMax), new Max());
-            this.fetchLatency.add(new Meter(new Count(), metrics.metricInstance(metricsRegistry.fetchRequestRate),
+            this.fetchLatency.add(new Meter(new WindowedCount(), metrics.metricInstance(metricsRegistry.fetchRequestRate),
                     metrics.metricInstance(metricsRegistry.fetchRequestTotal)));
 
             this.recordsFetchLag = metrics.sensor("records-lag");

clients/src/main/java/org/apache/kafka/common/metrics/MeasurableStat.java

@@ -19,7 +19,7 @@
 /**
  * A MeasurableStat is a {@link Stat} that is also {@link Measurable} (i.e. can produce a single floating point value).
  * This is the interface used for most of the simple statistics such as {@link org.apache.kafka.common.metrics.stats.Avg},
- * {@link org.apache.kafka.common.metrics.stats.Max}, {@link org.apache.kafka.common.metrics.stats.Count}, etc.
+ * {@link org.apache.kafka.common.metrics.stats.Max}, {@link org.apache.kafka.common.metrics.stats.CumulativeCount}, etc.
  */
 public interface MeasurableStat extends Stat, Measurable {
 

clients/src/main/java/org/apache/kafka/common/metrics/stats/Count.java

@@ -16,30 +16,14 @@
  */
 package org.apache.kafka.common.metrics.stats;
 
-import java.util.List;
-
-import org.apache.kafka.common.metrics.MetricConfig;
-
 /**
  * A {@link SampledStat} that maintains a simple count of what it has seen.
+ * This is a special kind of {@link WindowedSum} that always records a value of {@code 1} instead of the provided value.
+ *
+ * See also {@link CumulativeCount} for a non-sampled version of this metric.
+ *
+ * @deprecated since 2.4 . Use {@link WindowedCount} instead
  */
-public class Count extends SampledStat {
-
-    public Count() {
-        super(0);
-    }
-
-    @Override
-    protected void update(Sample sample, MetricConfig config, double value, long now) {
-        sample.value += 1.0;
-    }
-
-    @Override
-    public double combine(List<Sample> samples, MetricConfig config, long now) {
-        double total = 0.0;
-        for (Sample sample : samples)
-            total += sample.value;
-        return total;
-    }
-
+@Deprecated
+public class Count extends WindowedCount {
 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/CumulativeCount.java → clients/src/main/java/org/apache/kafka/common/metrics/stats/CumulativeCount.java

@@ -14,25 +14,21 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.kafka.streams.processor.internals.metrics;
+package org.apache.kafka.common.metrics.stats;
 
-import org.apache.kafka.common.metrics.MeasurableStat;
 import org.apache.kafka.common.metrics.MetricConfig;
 
 /**
- * A non-SampledStat version of Count for measuring -total metrics in streams
+ * A non-sampled version of {@link WindowedCount} maintained over all time.
+ *
+ * This is a special kind of {@link CumulativeSum} that always records {@code 1} instead of the provided value.
+ * In other words, it counts the number of
+ * {@link CumulativeCount#record(MetricConfig, double, long)} invocations,
+ * instead of summing the recorded values.
  */
-public class CumulativeCount implements MeasurableStat {
-
-    private double count = 0.0;
-
+public class CumulativeCount extends CumulativeSum {
     @Override
     public void record(final MetricConfig config, final double value, final long timeMs) {
-        count += 1;
-    }
-
-    @Override
-    public double measure(final MetricConfig config, final long now) {
-        return count;
+        super.record(config, 1, timeMs);
     }
 }

clients/src/main/java/org/apache/kafka/common/metrics/stats/CumulativeSum.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.kafka.common.metrics.stats;
+
+import org.apache.kafka.common.metrics.MeasurableStat;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+/**
+ * An non-sampled cumulative total maintained over all time.
+ * This is a non-sampled version of {@link WindowedSum}.
+ *
+ * See also {@link CumulativeCount} if you just want to increment the value by 1 on each recording.
+ */
+public class CumulativeSum implements MeasurableStat {
+
+    private double total;
+
+    public CumulativeSum() {
+        total = 0.0;
+    }
+
+    public CumulativeSum(double value) {
+        total = value;
+    }
+
+    @Override
+    public void record(MetricConfig config, double value, long now) {
+        total += value;
+    }
+
+    @Override
+    public double measure(MetricConfig config, long now) {
+        return total;
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java

@@ -23,48 +23,46 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.CompoundStat;
 import org.apache.kafka.common.metrics.MetricConfig;
-import org.apache.kafka.common.metrics.stats.Rate.SampledTotal;
 
 
 /**
  * A compound stat that includes a rate metric and a cumulative total metric.
  */
 public class Meter implements CompoundStat {
-
     private final MetricName rateMetricName;
     private final MetricName totalMetricName;
     private final Rate rate;
-    private final Total total;
+    private final CumulativeSum total;
 
     /**
-     * Construct a Meter with seconds as time unit and {@link SampledTotal} stats for Rate
+     * Construct a Meter with seconds as time unit
      */
     public Meter(MetricName rateMetricName, MetricName totalMetricName) {
-        this(TimeUnit.SECONDS, new SampledTotal(), rateMetricName, totalMetricName);
+        this(TimeUnit.SECONDS, new WindowedSum(), rateMetricName, totalMetricName);
     }
 
     /**
-     * Construct a Meter with provided time unit and {@link SampledTotal} stats for Rate
+     * Construct a Meter with provided time unit
      */
     public Meter(TimeUnit unit, MetricName rateMetricName, MetricName totalMetricName) {
-        this(unit, new SampledTotal(), rateMetricName, totalMetricName);
+        this(unit, new WindowedSum(), rateMetricName, totalMetricName);
     }
 
     /**
-     * Construct a Meter with seconds as time unit and provided {@link SampledStat} stats for Rate
+     * Construct a Meter with seconds as time unit
      */
     public Meter(SampledStat rateStat, MetricName rateMetricName, MetricName totalMetricName) {
         this(TimeUnit.SECONDS, rateStat, rateMetricName, totalMetricName);
     }
 
     /**
-     * Construct a Meter with provided time unit and provided {@link SampledStat} stats for Rate
+     * Construct a Meter with provided time unit
      */
     public Meter(TimeUnit unit, SampledStat rateStat, MetricName rateMetricName, MetricName totalMetricName) {
-        if (!(rateStat instanceof SampledTotal) && !(rateStat instanceof Count)) {
-            throw new IllegalArgumentException("Meter is supported only for SampledTotal and Count");
+        if (!(rateStat instanceof WindowedSum)) {
+            throw new IllegalArgumentException("Meter is supported only for WindowedCount or WindowedSum.");
         }
-        this.total = new Total();
+        this.total = new CumulativeSum();
         this.rate = new Rate(unit, rateStat);
         this.rateMetricName = rateMetricName;
         this.totalMetricName = totalMetricName;
@@ -81,7 +79,7 @@ public List<NamedMeasurable> stats() {
     public void record(MetricConfig config, double value, long timeMs) {
         rate.record(config, value, timeMs);
         // Total metrics with Count stat should record 1.0 (as recorded in the count)
-        double totalValue = (rate.stat instanceof Count) ? 1.0 : value;
+        double totalValue = (rate.stat instanceof WindowedCount) ? 1.0 : value;
         total.record(config, totalValue, timeMs);
     }
 }

clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.common.metrics.stats;
 
-import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.TimeUnit;
 
@@ -40,7 +39,7 @@ public Rate() {
     }
 
     public Rate(TimeUnit unit) {
-        this(unit, new SampledTotal());
+        this(unit, new WindowedSum());
     }
 
     public Rate(SampledStat stat) {
@@ -115,24 +114,10 @@ private double convert(long timeMs) {
         }
     }
 
-    public static class SampledTotal extends SampledStat {
-
-        public SampledTotal() {
-            super(0.0d);
-        }
-
-        @Override
-        protected void update(Sample sample, MetricConfig config, double value, long timeMs) {
-            sample.value += value;
-        }
-
-        @Override
-        public double combine(List<Sample> samples, MetricConfig config, long now) {
-            double total = 0.0;
-            for (Sample sample : samples)
-                total += sample.value;
-            return total;
-        }
-
+    /**
+     * @deprecated since 2.4 Use {@link SampledTotal} instead.
+     */
+    @Deprecated
+    public static class SampledTotal extends WindowedSum {
     }
 }

clients/src/main/java/org/apache/kafka/common/metrics/stats/Sum.java

@@ -16,30 +16,14 @@
  */
 package org.apache.kafka.common.metrics.stats;
 
-import java.util.List;
-
-import org.apache.kafka.common.metrics.MetricConfig;
-
 /**
  * A {@link SampledStat} that maintains the sum of what it has seen.
+ * This is a sampled version of {@link CumulativeSum}.
+ *
+ * See also {@link WindowedCount} if you want to increment the value by 1 on each recording.
+ *
+ * @deprecated since 2.4 . Use {@link WindowedSum} instead
  */
-public class Sum extends SampledStat {
-
-    public Sum() {
-        super(0);
-    }
-
-    @Override
-    protected void update(Sample sample, MetricConfig config, double value, long now) {
-        sample.value += value;
-    }
-
-    @Override
-    public double combine(List<Sample> samples, MetricConfig config, long now) {
-        double total = 0.0;
-        for (Sample sample : samples)
-            total += sample.value;
-        return total;
-    }
-
+@Deprecated
+public class Sum extends WindowedSum {
 }

clients/src/main/java/org/apache/kafka/common/metrics/stats/Total.java

@@ -16,32 +16,14 @@
  */
 package org.apache.kafka.common.metrics.stats;
 
-import org.apache.kafka.common.metrics.MeasurableStat;
-import org.apache.kafka.common.metrics.MetricConfig;
-
 /**
- * An un-windowed cumulative total maintained over all time.
+ * An non-sampled cumulative total maintained over all time.
+ * This is a non-sampled version of {@link WindowedSum}.
+ *
+ * See also {@link CumulativeCount} if you just want to increment the value by 1 on each recording.
+ *
+ * @deprecated since 2.4 . Use {@link CumulativeSum} instead.
  */
-public class Total implements MeasurableStat {
-
-    private double total;
-
-    public Total() {
-        this.total = 0.0;
-    }
-
-    public Total(double value) {
-        this.total = value;
-    }
-
-    @Override
-    public void record(MetricConfig config, double value, long now) {
-        this.total += value;
-    }
-
-    @Override
-    public double measure(MetricConfig config, long now) {
-        return this.total;
-    }
-
-}
+@Deprecated
+public class Total extends CumulativeSum {
+}