From a1172cb2e2c1d01356b902d59c4d218f81327ddd Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Wed, 13 May 2015 15:12:23 -0700 Subject: [PATCH] KAFKA-2191; Measured rate should not be infinite --- .../apache/kafka/common/metrics/stats/Rate.java | 2 ++ .../kafka/common/metrics/stats/SampledStat.java | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) 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 98429da..422a2be 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 @@ -59,6 +59,8 @@ public class Rate implements MeasurableStat { public double measure(MetricConfig config, long now) { double value = stat.measure(config, now); double elapsed = convert(now - stat.oldest(now).lastWindowMs); + if (elapsed == 0) + return 0; return value / elapsed; } diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java index b341b7d..2fd5ec2 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java +++ b/clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java @@ -42,8 +42,8 @@ public abstract class SampledStat implements MeasurableStat { @Override public void record(MetricConfig config, double value, long timeMs) { Sample sample = current(timeMs); - if (sample.isComplete(timeMs, config)) - sample = advance(config, timeMs); + while (sample.isComplete(timeMs, config)) + sample = advance(config, sample.lastWindowMs + config.timeWindowMs()); update(sample, config, value, timeMs); sample.eventCount += 1; } -- 1.7.9.5