diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index d5faa6e79dabc..b92a4a6109107 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/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 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, diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 08ab9fbbc5084..d4d028d38dc73 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/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"); diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/MeasurableStat.java b/clients/src/main/java/org/apache/kafka/common/metrics/MeasurableStat.java index aedac9a764c82..035449e7363a2 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/MeasurableStat.java +++ b/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 { diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Count.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Count.java index 3da91c4e908c6..2bef7cf93d8bb 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Count.java +++ b/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 samples, MetricConfig config, long now) { - double total = 0.0; - for (Sample sample : samples) - total += sample.value; - return total; - } - +@Deprecated +public class Count extends WindowedCount { } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/CumulativeCount.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/CumulativeCount.java similarity index 66% rename from streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/CumulativeCount.java rename to clients/src/main/java/org/apache/kafka/common/metrics/stats/CumulativeCount.java index 2c12c2b6e9d41..85591b5cf726f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/CumulativeCount.java +++ b/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); } } diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/CumulativeSum.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/CumulativeSum.java new file mode 100644 index 0000000000000..13f12a1bb09d4 --- /dev/null +++ b/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; + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java index 91d4461d2b52f..a6bdc9f3c108f 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java +++ b/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 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); } } diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java index a56734cd86dfb..604f860ae4f2d 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java +++ b/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 samples, MetricConfig config, long now) { - double total = 0.0; - for (Sample sample : samples) - total += sample.value; - return total; - } - + /** + * @deprecated since 2.4 Use {@link WindowedSum} instead. + */ + @Deprecated + public static class SampledTotal extends WindowedSum { } } diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Sum.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Sum.java index b40e9cdb649a0..17188b835c434 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Sum.java +++ b/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 samples, MetricConfig config, long now) { - double total = 0.0; - for (Sample sample : samples) - total += sample.value; - return total; - } - +@Deprecated +public class Sum extends WindowedSum { } diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Total.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Total.java index b8a83f5004bd6..23f7d040f8f90 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Total.java +++ b/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 { +} \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/WindowedCount.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/WindowedCount.java new file mode 100644 index 0000000000000..825f404ef02d1 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/metrics/stats/WindowedCount.java @@ -0,0 +1,35 @@ +/* + * 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.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. + * In other words, it counts the number of + * {@link WindowedCount#record(MetricConfig, double, long)} invocations, + * instead of summing the recorded values. + * + * See also {@link CumulativeCount} for a non-sampled version of this metric. + */ +public class WindowedCount extends WindowedSum { + @Override + protected void update(Sample sample, MetricConfig config, double value, long now) { + super.update(sample, config, 1.0, now); + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/WindowedSum.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/WindowedSum.java new file mode 100644 index 0000000000000..14aa5620fcc44 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/metrics/stats/WindowedSum.java @@ -0,0 +1,48 @@ +/* + * 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.MetricConfig; + +import java.util.List; + +/** + * 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. + */ +public class WindowedSum extends SampledStat { + + public WindowedSum() { + super(0); + } + + @Override + protected void update(Sample sample, MetricConfig config, double value, long now) { + sample.value += value; + } + + @Override + public double combine(List samples, MetricConfig config, long now) { + double total = 0.0; + for (Sample sample : samples) + total += sample.value; + return total; + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java index 20e24b7be4a2c..2652906099d78 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java +++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java @@ -23,11 +23,11 @@ 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.CumulativeSum; 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.metrics.stats.SampledStat; -import org.apache.kafka.common.metrics.stats.Total; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; @@ -1084,7 +1084,7 @@ public SelectorMetrics(Metrics metrics, String metricGrpPrefix, Map metricValueFunc) { s.add(metrics.metricName("test.min", "grp1"), new Min()); s.add(new Meter(TimeUnit.SECONDS, metrics.metricName("test.rate", "grp1"), metrics.metricName("test.total", "grp1"))); - s.add(new Meter(TimeUnit.SECONDS, new Count(), metrics.metricName("test.occurences", "grp1"), + s.add(new Meter(TimeUnit.SECONDS, new WindowedCount(), metrics.metricName("test.occurences", "grp1"), metrics.metricName("test.occurences.total", "grp1"))); - s.add(metrics.metricName("test.count", "grp1"), new Count()); + s.add(metrics.metricName("test.count", "grp1"), new WindowedCount()); s.add(new Percentiles(100, -100, 100, BucketSizing.CONSTANT, new Percentile(metrics.metricName("test.median", "grp1"), 50.0), new Percentile(metrics.metricName("test.perc99_9", "grp1"), 99.9))); Sensor s2 = metrics.sensor("test.sensor2"); - s2.add(metrics.metricName("s2.total", "grp1"), new Total()); + s2.add(metrics.metricName("s2.total", "grp1"), new CumulativeSum()); s2.record(5.0); int sum = 0; @@ -162,15 +162,15 @@ private void verifyStats(Function metricValueFunc) { @Test public void testHierarchicalSensors() { Sensor parent1 = metrics.sensor("test.parent1"); - parent1.add(metrics.metricName("test.parent1.count", "grp1"), new Count()); + parent1.add(metrics.metricName("test.parent1.count", "grp1"), new WindowedCount()); Sensor parent2 = metrics.sensor("test.parent2"); - parent2.add(metrics.metricName("test.parent2.count", "grp1"), new Count()); + parent2.add(metrics.metricName("test.parent2.count", "grp1"), new WindowedCount()); Sensor child1 = metrics.sensor("test.child1", parent1, parent2); - child1.add(metrics.metricName("test.child1.count", "grp1"), new Count()); + child1.add(metrics.metricName("test.child1.count", "grp1"), new WindowedCount()); Sensor child2 = metrics.sensor("test.child2", parent1); - child2.add(metrics.metricName("test.child2.count", "grp1"), new Count()); + child2.add(metrics.metricName("test.child2.count", "grp1"), new WindowedCount()); Sensor grandchild = metrics.sensor("test.grandchild", child1); - grandchild.add(metrics.metricName("test.grandchild.count", "grp1"), new Count()); + grandchild.add(metrics.metricName("test.grandchild.count", "grp1"), new WindowedCount()); /* increment each sensor one time */ parent1.record(); @@ -222,15 +222,15 @@ public void testRemoveChildSensor() { public void testRemoveSensor() { int size = metrics.metrics().size(); Sensor parent1 = metrics.sensor("test.parent1"); - parent1.add(metrics.metricName("test.parent1.count", "grp1"), new Count()); + parent1.add(metrics.metricName("test.parent1.count", "grp1"), new WindowedCount()); Sensor parent2 = metrics.sensor("test.parent2"); - parent2.add(metrics.metricName("test.parent2.count", "grp1"), new Count()); + parent2.add(metrics.metricName("test.parent2.count", "grp1"), new WindowedCount()); Sensor child1 = metrics.sensor("test.child1", parent1, parent2); - child1.add(metrics.metricName("test.child1.count", "grp1"), new Count()); + child1.add(metrics.metricName("test.child1.count", "grp1"), new WindowedCount()); Sensor child2 = metrics.sensor("test.child2", parent2); - child2.add(metrics.metricName("test.child2.count", "grp1"), new Count()); + child2.add(metrics.metricName("test.child2.count", "grp1"), new WindowedCount()); Sensor grandChild1 = metrics.sensor("test.gchild2", child2); - grandChild1.add(metrics.metricName("test.gchild2.count", "grp1"), new Count()); + grandChild1.add(metrics.metricName("test.gchild2.count", "grp1"), new WindowedCount()); Sensor sensor = metrics.getSensor("test.parent1"); assertNotNull(sensor); @@ -268,10 +268,10 @@ public void testRemoveSensor() { @Test public void testRemoveInactiveMetrics() { Sensor s1 = metrics.sensor("test.s1", null, 1); - s1.add(metrics.metricName("test.s1.count", "grp1"), new Count()); + s1.add(metrics.metricName("test.s1.count", "grp1"), new WindowedCount()); Sensor s2 = metrics.sensor("test.s2", null, 3); - s2.add(metrics.metricName("test.s2.count", "grp1"), new Count()); + s2.add(metrics.metricName("test.s2.count", "grp1"), new WindowedCount()); Metrics.ExpireSensorTask purger = metrics.new ExpireSensorTask(); purger.run(); @@ -309,7 +309,7 @@ public void testRemoveInactiveMetrics() { // After purging, it should be possible to recreate a metric s1 = metrics.sensor("test.s1", null, 1); - s1.add(metrics.metricName("test.s1.count", "grp1"), new Count()); + s1.add(metrics.metricName("test.s1.count", "grp1"), new WindowedCount()); assertNotNull("Sensor test.s1 must be present", metrics.getSensor("test.s1")); assertNotNull("MetricName test.s1.count must be present", metrics.metrics().get(metrics.metricName("test.s1.count", "grp1"))); @@ -318,8 +318,8 @@ public void testRemoveInactiveMetrics() { @Test public void testRemoveMetric() { int size = metrics.metrics().size(); - metrics.addMetric(metrics.metricName("test1", "grp1"), new Count()); - metrics.addMetric(metrics.metricName("test2", "grp1"), new Count()); + metrics.addMetric(metrics.metricName("test1", "grp1"), new WindowedCount()); + metrics.addMetric(metrics.metricName("test2", "grp1"), new WindowedCount()); assertNotNull(metrics.removeMetric(metrics.metricName("test1", "grp1"))); assertNull(metrics.metrics().get(metrics.metricName("test1", "grp1"))); @@ -333,7 +333,7 @@ public void testRemoveMetric() { @Test public void testEventWindowing() { - Count count = new Count(); + WindowedCount count = new WindowedCount(); MetricConfig config = new MetricConfig().eventWindow(1).samples(2); count.record(config, 1.0, time.milliseconds()); count.record(config, 1.0, time.milliseconds()); @@ -344,7 +344,7 @@ public void testEventWindowing() { @Test public void testTimeWindowing() { - Count count = new Count(); + WindowedCount count = new WindowedCount(); MetricConfig config = new MetricConfig().timeWindow(1, TimeUnit.MILLISECONDS).samples(2); count.record(config, 1.0, time.milliseconds()); time.sleep(1); @@ -397,8 +397,8 @@ public void testSampledStatReturnsNaNWhenNoValuesExist() { */ @Test public void testSampledStatReturnsInitialValueWhenNoValuesExist() { - Count count = new Count(); - Rate.SampledTotal sampledTotal = new Rate.SampledTotal(); + WindowedCount count = new WindowedCount(); + WindowedSum sampledTotal = new WindowedSum(); long windowMs = 100; int samples = 2; MetricConfig config = new MetricConfig().timeWindow(windowMs, TimeUnit.MILLISECONDS).samples(samples); @@ -415,14 +415,14 @@ public void testSampledStatReturnsInitialValueWhenNoValuesExist() { @Test(expected = IllegalArgumentException.class) public void testDuplicateMetricName() { metrics.sensor("test").add(metrics.metricName("test", "grp1"), new Avg()); - metrics.sensor("test2").add(metrics.metricName("test", "grp1"), new Total()); + metrics.sensor("test2").add(metrics.metricName("test", "grp1"), new CumulativeSum()); } @Test public void testQuotas() { Sensor sensor = metrics.sensor("test"); - sensor.add(metrics.metricName("test1.total", "grp1"), new Total(), new MetricConfig().quota(Quota.upperBound(5.0))); - sensor.add(metrics.metricName("test2.total", "grp1"), new Total(), new MetricConfig().quota(Quota.lowerBound(0.0))); + sensor.add(metrics.metricName("test1.total", "grp1"), new CumulativeSum(), new MetricConfig().quota(Quota.upperBound(5.0))); + sensor.add(metrics.metricName("test2.total", "grp1"), new CumulativeSum(), new MetricConfig().quota(Quota.lowerBound(0.0))); sensor.record(5.0); try { sensor.record(1.0); @@ -503,7 +503,7 @@ public void testRateWindowing() throws Exception { MetricName countRateMetricName = metrics.metricName("test.count.rate", "grp1"); MetricName countTotalMetricName = metrics.metricName("test.count.total", "grp1"); s.add(new Meter(TimeUnit.SECONDS, rateMetricName, totalMetricName)); - s.add(new Meter(TimeUnit.SECONDS, new Count(), countRateMetricName, countTotalMetricName)); + s.add(new Meter(TimeUnit.SECONDS, new WindowedCount(), countRateMetricName, countTotalMetricName)); KafkaMetric totalMetric = metrics.metrics().get(totalMetricName); KafkaMetric countTotalMetric = metrics.metrics().get(countTotalMetricName); @@ -825,10 +825,10 @@ private Sensor createSensor(StatType statType, int index) { sensor.add(metrics.metricName("test.metric.avg", "avg", tags), new Avg()); break; case TOTAL: - sensor.add(metrics.metricName("test.metric.total", "total", tags), new Total()); + sensor.add(metrics.metricName("test.metric.total", "total", tags), new CumulativeSum()); break; case COUNT: - sensor.add(metrics.metricName("test.metric.count", "count", tags), new Count()); + sensor.add(metrics.metricName("test.metric.count", "count", tags), new WindowedCount()); break; case MAX: sensor.add(metrics.metricName("test.metric.max", "max", tags), new Max()); @@ -843,7 +843,7 @@ private Sensor createSensor(StatType statType, int index) { sensor.add(metrics.metricName("test.metric.simpleRate", "simpleRate", tags), new SimpleRate()); break; case SUM: - sensor.add(metrics.metricName("test.metric.sum", "sum", tags), new Sum()); + sensor.add(metrics.metricName("test.metric.sum", "sum", tags), new WindowedSum()); break; case VALUE: sensor.add(metrics.metricName("test.metric.value", "value", tags), new Value()); diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java index 8e3dfeb3b9fad..fc7cfe24def9a 100644 --- a/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java @@ -20,7 +20,7 @@ import org.apache.kafka.common.metrics.stats.Avg; import org.apache.kafka.common.metrics.stats.Meter; import org.apache.kafka.common.metrics.stats.Rate; -import org.apache.kafka.common.metrics.stats.Sum; +import org.apache.kafka.common.metrics.stats.WindowedSum; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.SystemTime; import org.apache.kafka.common.utils.Time; @@ -128,7 +128,7 @@ public void testIdempotentAdd() { } // note that adding a different metric with the same name is also a no-op - assertTrue(sensor.add(metrics.metricName("test-metric", "test-group"), new Sum())); + assertTrue(sensor.add(metrics.metricName("test-metric", "test-group"), new WindowedSum())); // so after all this, we still just have the original metric registered assertEquals(1, sensor.metrics().size()); diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/stats/MeterTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/stats/MeterTest.java index 8204771d794db..27198ea1e79d0 100644 --- a/clients/src/test/java/org/apache/kafka/common/metrics/stats/MeterTest.java +++ b/clients/src/test/java/org/apache/kafka/common/metrics/stats/MeterTest.java @@ -44,7 +44,7 @@ public void testMeter() { assertEquals(rateMetricName, rate.name()); assertEquals(totalMetricName, total.name()); Rate rateStat = (Rate) rate.stat(); - Total totalStat = (Total) total.stat(); + CumulativeSum totalStat = (CumulativeSum) total.stat(); MetricConfig config = new MetricConfig(); double nextValue = 0.0; diff --git a/clients/src/test/java/org/apache/kafka/test/MetricsBench.java b/clients/src/test/java/org/apache/kafka/test/MetricsBench.java index 379db2f526d1d..93cbf6d2a8910 100644 --- a/clients/src/test/java/org/apache/kafka/test/MetricsBench.java +++ b/clients/src/test/java/org/apache/kafka/test/MetricsBench.java @@ -21,11 +21,11 @@ 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.Percentile; import org.apache.kafka.common.metrics.stats.Percentiles; import org.apache.kafka.common.metrics.stats.Percentiles.BucketSizing; +import org.apache.kafka.common.metrics.stats.WindowedCount; public class MetricsBench { @@ -37,7 +37,7 @@ public static void main(String[] args) { Sensor child = metrics.sensor("child", parent); for (Sensor sensor : Arrays.asList(parent, child)) { sensor.add(metrics.metricName(sensor.name() + ".avg", "grp1"), new Avg()); - sensor.add(metrics.metricName(sensor.name() + ".count", "grp1"), new Count()); + sensor.add(metrics.metricName(sensor.name() + ".count", "grp1"), new WindowedCount()); sensor.add(metrics.metricName(sensor.name() + ".max", "grp1"), new Max()); sensor.add(new Percentiles(1024, 0.0, diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java index f848a18f06f9e..fd90dd149ba51 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java @@ -24,8 +24,8 @@ import org.apache.kafka.common.config.ConfigValue; import org.apache.kafka.common.config.provider.ConfigProvider; import org.apache.kafka.common.metrics.Sensor; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.metrics.stats.Frequencies; -import org.apache.kafka.common.metrics.stats.Total; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.connect.connector.Connector; @@ -863,13 +863,13 @@ public Double metricValue(long now) { connectorStartupResults.add(connectorStartupResultFrequencies); connectorStartupAttempts = metricGroup.sensor("connector-startup-attempts"); - connectorStartupAttempts.add(metricGroup.metricName(registry.connectorStartupAttemptsTotal), new Total()); + connectorStartupAttempts.add(metricGroup.metricName(registry.connectorStartupAttemptsTotal), new CumulativeSum()); connectorStartupSuccesses = metricGroup.sensor("connector-startup-successes"); - connectorStartupSuccesses.add(metricGroup.metricName(registry.connectorStartupSuccessTotal), new Total()); + connectorStartupSuccesses.add(metricGroup.metricName(registry.connectorStartupSuccessTotal), new CumulativeSum()); connectorStartupFailures = metricGroup.sensor("connector-startup-failures"); - connectorStartupFailures.add(metricGroup.metricName(registry.connectorStartupFailureTotal), new Total()); + connectorStartupFailures.add(metricGroup.metricName(registry.connectorStartupFailureTotal), new CumulativeSum()); MetricName taskFailurePct = metricGroup.metricName(registry.taskStartupFailurePercentage); MetricName taskSuccessPct = metricGroup.metricName(registry.taskStartupSuccessPercentage); @@ -878,13 +878,13 @@ public Double metricValue(long now) { taskStartupResults.add(taskStartupResultFrequencies); taskStartupAttempts = metricGroup.sensor("task-startup-attempts"); - taskStartupAttempts.add(metricGroup.metricName(registry.taskStartupAttemptsTotal), new Total()); + taskStartupAttempts.add(metricGroup.metricName(registry.taskStartupAttemptsTotal), new CumulativeSum()); taskStartupSuccesses = metricGroup.sensor("task-startup-successes"); - taskStartupSuccesses.add(metricGroup.metricName(registry.taskStartupSuccessTotal), new Total()); + taskStartupSuccesses.add(metricGroup.metricName(registry.taskStartupSuccessTotal), new CumulativeSum()); taskStartupFailures = metricGroup.sensor("task-startup-failures"); - taskStartupFailures.add(metricGroup.metricName(registry.taskStartupFailureTotal), new Total()); + taskStartupFailures.add(metricGroup.metricName(registry.taskStartupFailureTotal), new CumulativeSum()); } void close() { diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java index f21c500699e07..395c93e0f772a 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java @@ -27,9 +27,9 @@ import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.metrics.stats.Rate; -import org.apache.kafka.common.metrics.stats.Total; import org.apache.kafka.common.metrics.stats.Value; import org.apache.kafka.common.utils.Time; import org.apache.kafka.connect.data.SchemaAndValue; @@ -705,11 +705,11 @@ public SinkTaskMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) { sinkRecordRead = metricGroup.sensor("sink-record-read"); sinkRecordRead.add(metricGroup.metricName(registry.sinkRecordReadRate), new Rate()); - sinkRecordRead.add(metricGroup.metricName(registry.sinkRecordReadTotal), new Total()); + sinkRecordRead.add(metricGroup.metricName(registry.sinkRecordReadTotal), new CumulativeSum()); sinkRecordSend = metricGroup.sensor("sink-record-send"); sinkRecordSend.add(metricGroup.metricName(registry.sinkRecordSendRate), new Rate()); - sinkRecordSend.add(metricGroup.metricName(registry.sinkRecordSendTotal), new Total()); + sinkRecordSend.add(metricGroup.metricName(registry.sinkRecordSendTotal), new CumulativeSum()); sinkRecordActiveCount = metricGroup.sensor("sink-record-active-count"); sinkRecordActiveCount.add(metricGroup.metricName(registry.sinkRecordActiveCount), new Value()); @@ -724,11 +724,11 @@ public SinkTaskMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) { offsetCompletion = metricGroup.sensor("offset-commit-completion"); offsetCompletion.add(metricGroup.metricName(registry.sinkRecordOffsetCommitCompletionRate), new Rate()); - offsetCompletion.add(metricGroup.metricName(registry.sinkRecordOffsetCommitCompletionTotal), new Total()); + offsetCompletion.add(metricGroup.metricName(registry.sinkRecordOffsetCommitCompletionTotal), new CumulativeSum()); offsetCompletionSkip = metricGroup.sensor("offset-commit-completion-skip"); offsetCompletionSkip.add(metricGroup.metricName(registry.sinkRecordOffsetCommitSkipRate), new Rate()); - offsetCompletionSkip.add(metricGroup.metricName(registry.sinkRecordOffsetCommitSkipTotal), new Total()); + offsetCompletionSkip.add(metricGroup.metricName(registry.sinkRecordOffsetCommitSkipTotal), new CumulativeSum()); putBatchTime = metricGroup.sensor("put-batch-time"); putBatchTime.add(metricGroup.metricName(registry.sinkRecordPutBatchTimeMax), new Max()); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java index 6e94c6f2e4b0d..6e1152be91168 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java @@ -24,9 +24,9 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.metrics.stats.Rate; -import org.apache.kafka.common.metrics.stats.Total; import org.apache.kafka.common.metrics.stats.Value; import org.apache.kafka.common.utils.Time; import org.apache.kafka.connect.errors.ConnectException; @@ -591,11 +591,11 @@ public SourceTaskMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) sourceRecordPoll = metricGroup.sensor("source-record-poll"); sourceRecordPoll.add(metricGroup.metricName(registry.sourceRecordPollRate), new Rate()); - sourceRecordPoll.add(metricGroup.metricName(registry.sourceRecordPollTotal), new Total()); + sourceRecordPoll.add(metricGroup.metricName(registry.sourceRecordPollTotal), new CumulativeSum()); sourceRecordWrite = metricGroup.sensor("source-record-write"); sourceRecordWrite.add(metricGroup.metricName(registry.sourceRecordWriteRate), new Rate()); - sourceRecordWrite.add(metricGroup.metricName(registry.sourceRecordWriteTotal), new Total()); + sourceRecordWrite.add(metricGroup.metricName(registry.sourceRecordWriteTotal), new CumulativeSum()); pollTime = metricGroup.sensor("poll-batch-time"); pollTime.add(metricGroup.metricName(registry.sourceRecordPollBatchTimeMax), new Max()); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index b1a41c0b4ba64..86caeeb4af6f1 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -22,8 +22,8 @@ import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.metrics.stats.Max; -import org.apache.kafka.common.metrics.stats.Total; import org.apache.kafka.common.utils.Exit; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; @@ -1483,7 +1483,7 @@ public Double metricValue(long now) { }); rebalanceCompletedCounts = metricGroup.sensor("completed-rebalance-count"); - rebalanceCompletedCounts.add(metricGroup.metricName(registry.rebalanceCompletedTotal), new Total()); + rebalanceCompletedCounts.add(metricGroup.metricName(registry.rebalanceCompletedTotal), new CumulativeSum()); rebalanceTime = metricGroup.sensor("rebalance-time"); rebalanceTime.add(metricGroup.metricName(registry.rebalanceTimeMax), new Max()); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorHandlingMetrics.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorHandlingMetrics.java index c589012d02cc2..0deecd129a06e 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorHandlingMetrics.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorHandlingMetrics.java @@ -17,7 +17,7 @@ package org.apache.kafka.connect.runtime.errors; import org.apache.kafka.common.metrics.Sensor; -import org.apache.kafka.common.metrics.stats.Total; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.utils.SystemTime; import org.apache.kafka.common.utils.Time; import org.apache.kafka.connect.runtime.ConnectMetrics; @@ -62,25 +62,25 @@ public ErrorHandlingMetrics(ConnectorTaskId id, ConnectMetrics connectMetrics) { metricGroup.close(); recordProcessingFailures = metricGroup.sensor("total-record-failures"); - recordProcessingFailures.add(metricGroup.metricName(registry.recordProcessingFailures), new Total()); + recordProcessingFailures.add(metricGroup.metricName(registry.recordProcessingFailures), new CumulativeSum()); recordProcessingErrors = metricGroup.sensor("total-record-errors"); - recordProcessingErrors.add(metricGroup.metricName(registry.recordProcessingErrors), new Total()); + recordProcessingErrors.add(metricGroup.metricName(registry.recordProcessingErrors), new CumulativeSum()); recordsSkipped = metricGroup.sensor("total-records-skipped"); - recordsSkipped.add(metricGroup.metricName(registry.recordsSkipped), new Total()); + recordsSkipped.add(metricGroup.metricName(registry.recordsSkipped), new CumulativeSum()); retries = metricGroup.sensor("total-retries"); - retries.add(metricGroup.metricName(registry.retries), new Total()); + retries.add(metricGroup.metricName(registry.retries), new CumulativeSum()); errorsLogged = metricGroup.sensor("total-errors-logged"); - errorsLogged.add(metricGroup.metricName(registry.errorsLogged), new Total()); + errorsLogged.add(metricGroup.metricName(registry.errorsLogged), new CumulativeSum()); dlqProduceRequests = metricGroup.sensor("deadletterqueue-produce-requests"); - dlqProduceRequests.add(metricGroup.metricName(registry.dlqProduceRequests), new Total()); + dlqProduceRequests.add(metricGroup.metricName(registry.dlqProduceRequests), new CumulativeSum()); dlqProduceFailures = metricGroup.sensor("deadletterqueue-produce-failures"); - dlqProduceFailures.add(metricGroup.metricName(registry.dlqProduceFailures), new Total()); + dlqProduceFailures.add(metricGroup.metricName(registry.dlqProduceFailures), new CumulativeSum()); metricGroup.addValueMetric(registry.lastErrorTimestamp, now -> lastErrorTime); } diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala index 619d2605ac9b4..69f02ae5799be 100644 --- a/core/src/main/scala/kafka/network/SocketServer.scala +++ b/core/src/main/scala/kafka/network/SocketServer.scala @@ -39,8 +39,7 @@ import org.apache.kafka.common.config.ConfigException import org.apache.kafka.common.{KafkaException, Reconfigurable} import org.apache.kafka.common.memory.{MemoryPool, SimpleMemoryPool} import org.apache.kafka.common.metrics._ -import org.apache.kafka.common.metrics.stats.Meter -import org.apache.kafka.common.metrics.stats.Total +import org.apache.kafka.common.metrics.stats.{CumulativeSum, Meter} import org.apache.kafka.common.network.KafkaChannel.ChannelMuteEvent import org.apache.kafka.common.network.{ChannelBuilder, ChannelBuilders, KafkaChannel, ListenerName, ListenerReconfigurable, Selectable, Send, Selector => KSelector} import org.apache.kafka.common.protocol.ApiKeys @@ -712,7 +711,7 @@ private[kafka] class Processor(val id: Int, Map(NetworkProcessorMetricTag -> id.toString) ) - val expiredConnectionsKilledCount = new Total() + val expiredConnectionsKilledCount = new CumulativeSum() private val expiredConnectionsKilledCountMetricName = metrics.metricName("expired-connections-killed-count", "socket-server-metrics", metricTags) metrics.addMetric(expiredConnectionsKilledCountMetricName, expiredConnectionsKilledCount) diff --git a/core/src/main/scala/kafka/server/ClientQuotaManager.scala b/core/src/main/scala/kafka/server/ClientQuotaManager.scala index 959144cf49901..5451b5caa35df 100644 --- a/core/src/main/scala/kafka/server/ClientQuotaManager.scala +++ b/core/src/main/scala/kafka/server/ClientQuotaManager.scala @@ -27,7 +27,7 @@ import kafka.utils.{Logging, ShutdownableThread} import org.apache.kafka.common.{Cluster, MetricName} import org.apache.kafka.common.metrics._ import org.apache.kafka.common.metrics.Metrics -import org.apache.kafka.common.metrics.stats.{Avg, Rate, Total} +import org.apache.kafka.common.metrics.stats.{Avg, CumulativeSum, Rate} import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.{Sanitizer, Time} import org.apache.kafka.server.quota.{ClientQuotaCallback, ClientQuotaEntity, ClientQuotaType} @@ -179,7 +179,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig, private val delayQueueSensor = metrics.sensor(quotaType + "-delayQueue") delayQueueSensor.add(metrics.metricName("queue-size", quotaType.toString, - "Tracks the size of the delay queue"), new Total()) + "Tracks the size of the delay queue"), new CumulativeSum()) start() // Use start method to keep spotbugs happy private def start() { throttledChannelReaper.start() diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/metrics/Sensors.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/metrics/Sensors.java index 4ecaeb4189755..038b8ac77e9a3 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/metrics/Sensors.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/metrics/Sensors.java @@ -19,10 +19,10 @@ import org.apache.kafka.common.MetricName; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.metrics.stats.Rate; -import org.apache.kafka.common.metrics.stats.Sum; -import org.apache.kafka.common.metrics.stats.Total; +import org.apache.kafka.common.metrics.stats.WindowedSum; import org.apache.kafka.streams.processor.internals.InternalProcessorContext; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; @@ -106,7 +106,7 @@ public static Sensor suppressionEmitSensor(final InternalProcessorContext contex "The average number of occurrence of suppression-emit operation per second.", tags ), - new Rate(TimeUnit.SECONDS, new Sum()) + new Rate(TimeUnit.SECONDS, new WindowedSum()) ); sensor.add( new MetricName( @@ -115,7 +115,7 @@ public static Sensor suppressionEmitSensor(final InternalProcessorContext contex "The total number of occurrence of suppression-emit operations.", tags ), - new Total() + new CumulativeSum() ); return sensor; } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index 210412be92fc6..836330f242617 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -28,9 +28,10 @@ import org.apache.kafka.common.errors.TimeoutException; 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.CumulativeCount; import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.metrics.stats.Rate; +import org.apache.kafka.common.metrics.stats.WindowedCount; import org.apache.kafka.common.utils.Time; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.DeserializationExceptionHandler; @@ -43,7 +44,6 @@ import org.apache.kafka.streams.processor.Punctuator; import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.TimestampExtractor; -import org.apache.kafka.streams.processor.internals.metrics.CumulativeCount; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; import org.apache.kafka.streams.processor.internals.metrics.ThreadMetrics; import org.apache.kafka.streams.state.internals.ThreadCache; @@ -112,7 +112,7 @@ protected static final class TaskMetrics { ); taskCommitTimeSensor.add( new MetricName("commit-rate", group, "The average number of occurrence of commit operation per second.", tagMap), - new Rate(TimeUnit.SECONDS, new Count()) + new Rate(TimeUnit.SECONDS, new WindowedCount()) ); taskCommitTimeSensor.add( new MetricName("commit-total", group, "The total number of occurrence of commit operations.", tagMap), @@ -123,7 +123,7 @@ protected static final class TaskMetrics { taskEnforcedProcessSensor = metrics.taskLevelSensor(taskName, "enforced-processing", Sensor.RecordingLevel.DEBUG, parent); taskEnforcedProcessSensor.add( new MetricName("enforced-processing-rate", group, "The average number of occurrence of enforced-processing operation per second.", tagMap), - new Rate(TimeUnit.SECONDS, new Count()) + new Rate(TimeUnit.SECONDS, new WindowedCount()) ); taskEnforcedProcessSensor.add( new MetricName("enforced-processing-total", group, "The total number of occurrence of enforced-processing operations.", tagMap), diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java index 46b5669d6011c..b6bfcc5c8513a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java @@ -22,9 +22,10 @@ import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.Sensor.RecordingLevel; import org.apache.kafka.common.metrics.stats.Avg; -import org.apache.kafka.common.metrics.stats.Count; +import org.apache.kafka.common.metrics.stats.CumulativeCount; import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.metrics.stats.Rate; +import org.apache.kafka.common.metrics.stats.WindowedCount; import org.apache.kafka.streams.StreamsMetrics; import java.util.Arrays; @@ -445,7 +446,7 @@ public static void addInvocationRateAndCount(final Sensor sensor, descriptionOfRate, tags ), - new Rate(TimeUnit.SECONDS, new Count()) + new Rate(TimeUnit.SECONDS, new WindowedCount()) ); } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java index a7da2cbec76cb..47dd61b2a9057 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java @@ -33,7 +33,7 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; -import org.apache.kafka.common.metrics.stats.Sum; +import org.apache.kafka.common.metrics.stats.WindowedSum; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.LogContext; @@ -215,7 +215,7 @@ public void shouldRecordSkippedMetricAndLogWarningIfSendFailsWithContinueExcepti final Sensor sensor = metrics.sensor("skipped-records"); final LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(); final MetricName metricName = new MetricName("name", "group", "description", Collections.emptyMap()); - sensor.add(metricName, new Sum()); + sensor.add(metricName, new WindowedSum()); final RecordCollector collector = new RecordCollectorImpl( "test", logContext, diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java index 0400128282e3d..2faa078d6751e 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java @@ -27,7 +27,7 @@ import org.apache.kafka.common.metrics.KafkaMetric; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; -import org.apache.kafka.common.metrics.stats.Total; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.IntegerSerializer; @@ -692,7 +692,7 @@ void closeStateManager(final boolean clean) throws ProcessorStateException { private MetricName setupCloseTaskMetric() { final MetricName metricName = new MetricName("name", "group", "description", Collections.emptyMap()); final Sensor sensor = streamsMetrics.threadLevelSensor("task-closed", Sensor.RecordingLevel.INFO); - sensor.add(metricName, new Total()); + sensor.add(metricName, new CumulativeSum()); return metricName; } diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java index 48c038c2d5f65..2b1428f9f8968 100644 --- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java +++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java @@ -32,9 +32,9 @@ import org.apache.kafka.common.metrics.MetricConfig; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; -import org.apache.kafka.common.metrics.stats.Count; +import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.metrics.stats.Rate; -import org.apache.kafka.common.metrics.stats.Total; +import org.apache.kafka.common.metrics.stats.WindowedCount; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.Deserializer; @@ -289,12 +289,12 @@ public List partitionsFor(final String topic) { threadLevelGroup, "The average per-second number of skipped records", streamsMetrics.tagMap()), - new Rate(TimeUnit.SECONDS, new Count())); + new Rate(TimeUnit.SECONDS, new WindowedCount())); skippedRecordsSensor.add(new MetricName("skipped-records-total", threadLevelGroup, "The total number of skipped records", streamsMetrics.tagMap()), - new Total()); + new CumulativeSum()); final ThreadCache cache = new ThreadCache( new LogContext("topology-test-driver "), Math.max(0, streamsConfig.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG)),