MINOR: Fix rate metric spikes

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

@@ -63,33 +63,38 @@ public void record(MetricConfig config, double value, long timeMs) {
 
     @Override
     public double measure(MetricConfig config, long now) {
+        long windowSizeMs = windowSize(config, now);
         double value = stat.measure(config, now);
-        return value / convert(windowSize(config, now), unit);
+        return value / convert(windowSizeMs, unit);
     }
 
     public long windowSize(MetricConfig config, long now) {
         // purge old samples before we compute the window size
-        stat.purgeObsoleteSamples(config, now);
+        long purgedUpToMs = stat.purgeObsoleteSamples(config, now);
 
         /*
          * Here we check the total amount of time elapsed since the oldest non-obsolete window.
          * This give the total windowSize of the batch which is the time used for Rate computation.
-         * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30 second
+         * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30-second
          * window, the measured rate will be very high.
-         * Hence we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
+         * Hence, we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
          *
          * Note that we could simply count the amount of time elapsed in the current window and add n-1 windows to get the total time,
          * but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
          * if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
          */
-        long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
+        long totalElapsedTimeMs = now - stat.oldest(now).startTimeMs;
         // Check how many full windows of data we have currently retained
         int numFullWindows = (int) (totalElapsedTimeMs / config.timeWindowMs());
         int minFullWindows = config.samples() - 1;
 
         // If the available windows are less than the minimum required, add the difference to the totalElapsedTime
-        if (numFullWindows < minFullWindows)
+        if (numFullWindows < minFullWindows) {
             totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
+        }
+
+        // if some part of considered interval was just purged with its data, exclude it 
+        totalElapsedTimeMs = Math.min(totalElapsedTimeMs, now - purgedUpToMs);
 
         // If window size is being calculated at the exact beginning of the window with no prior samples, the window size
         // will result in a value of 0. Calculation of rate over a window is size 0 is undefined, hence, we assume the

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

@@ -87,7 +87,7 @@ public Sample oldest(long now) {
         Sample oldest = this.samples.get(0);
         for (int i = 1; i < this.samples.size(); i++) {
             Sample curr = this.samples.get(i);
-            if (curr.lastWindowMs < oldest.lastWindowMs)
+            if (curr.startTimeMs < oldest.startTimeMs)
                 oldest = curr;
         }
         return oldest;
@@ -106,44 +106,55 @@ public String toString() {
 
     public abstract double combine(List<Sample> samples, MetricConfig config, long now);
 
-    /* Timeout any windows that have expired in the absence of any events */
-    protected void purgeObsoleteSamples(MetricConfig config, long now) {
+    /**
+     * Purges any windows that started before the configured period.
+     * Returns the end of the latest purged window.
+     */
+    protected long purgeObsoleteSamples(MetricConfig config, long now) {
         long expireAge = config.samples() * config.timeWindowMs();
+        long purgedUpToMs = 0;
         for (Sample sample : samples) {
-            if (now - sample.lastWindowMs >= expireAge)
+            if (now - sample.startTimeMs >= expireAge) {
+                purgedUpToMs = Math.max(purgedUpToMs, sample.endTimeMs(config));
                 sample.reset(now);
+            }
         }
+        return purgedUpToMs;
     }
 
     protected static class Sample {
         public double initialValue;
         public long eventCount;
-        public long lastWindowMs;
+        public long startTimeMs;
         public double value;
 
         public Sample(double initialValue, long now) {
             this.initialValue = initialValue;
             this.eventCount = 0;
-            this.lastWindowMs = now;
+            this.startTimeMs = now;
             this.value = initialValue;
         }
 
         public void reset(long now) {
             this.eventCount = 0;
-            this.lastWindowMs = now;
+            this.startTimeMs = now;
             this.value = initialValue;
         }
 
         public boolean isComplete(long timeMs, MetricConfig config) {
-            return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+            return timeMs - startTimeMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+        }
+
+        public long endTimeMs(MetricConfig config) {
+            return startTimeMs + config.timeWindowMs();
         }
 
         @Override
         public String toString() {
             return "Sample(" +
                 "value=" + value +
                 ", eventCount=" + eventCount +
-                ", lastWindowMs=" + lastWindowMs +
+                ", startTimeMs=" + startTimeMs +
                 ", initialValue=" + initialValue +
                 ')';
         }

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

@@ -33,7 +33,7 @@ public class SimpleRate extends Rate {
     @Override
     public long windowSize(MetricConfig config, long now) {
         stat.purgeObsoleteSamples(config, now);
-        long elapsed = now - stat.oldest(now).lastWindowMs;
+        long elapsed = now - stat.oldest(now).startTimeMs;
         return Math.max(elapsed, config.timeWindowMs());
     }
 }

clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java

@@ -21,23 +21,26 @@
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.common.utils.Time;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.*;
 
 public class RateTest {
     private static final double EPS = 0.000001;
-    private Rate r;
-    private Time timeClock;
+    private Rate rate;
+    private Time time;
 
     @BeforeEach
     public void setup() {
-        r = new Rate();
-        timeClock = new MockTime();
+        rate = new Rate();
+        time = new MockTime();
     }
 
     // Tests the scenario where the recording and measurement is done before the window for first sample finishes
@@ -48,12 +51,12 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
         final MetricConfig config = new MetricConfig().samples(numSample).timeWindow(sampleWindowSizeSec, TimeUnit.SECONDS);
         final double sampleValue = 50.0;
         // record at beginning of the window
-        r.record(config, sampleValue, timeClock.milliseconds());
+        rate.record(config, sampleValue, time.milliseconds());
         // forward time till almost the end of window
         final long measurementTime = TimeUnit.SECONDS.toMillis(sampleWindowSizeSec) - 1;
-        timeClock.sleep(measurementTime);
+        time.sleep(measurementTime);
         // calculate rate at almost the end of window
-        final double observedRate = r.measure(config, timeClock.milliseconds());
+        final double observedRate = rate.measure(config, time.milliseconds());
         assertFalse(Double.isNaN(observedRate));
 
         // In a scenario where sufficient number of samples is not available yet, the rate calculation algorithm assumes
@@ -64,4 +67,31 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
         double expectedRatePerSec = sampleValue / windowSize;
         assertEquals(expectedRatePerSec, observedRate, EPS);
     }
+
+
+    // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained 
+    // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of 
+    // measurement. Start assertions from the second window.
+    @Test
+    public void testRateIsConsistentAfterTheFirstWindow() {
+        MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+        List<Integer> steps = Arrays.asList(0, 99, 100, 100, 100, 100, 100, 100, 100, 100, 100);
+
+        // start the first window and record events at 0,99,199,...,999 ms 
+        for (int stepMs : steps) {
+            time.sleep(stepMs);
+            rate.record(config, 1, time.milliseconds());
+        }
+
+        // making a gap of 100 ms between windows
+        time.sleep(101);
+
+        // start the second window and record events at 0,99,199,...,999 ms
+        for (int stepMs : steps) {
+            time.sleep(stepMs);
+            rate.record(config, 1, time.milliseconds());
+            double observedRate = rate.measure(config, time.milliseconds());
+            assertTrue(observedRate >= 10 && observedRate <= 11);
+        }
+    }
 }