Address PR comments

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

@@ -74,13 +74,16 @@ public long windowSize(MetricConfig config, long 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.
+         *
+         * Note also, that totalElapsedTimeMs can be larger than the monitored window size,
+         * if the oldest sample started before the window while overlapping it.
          */
         long totalElapsedTimeMs = now - stat.oldest(now).startTimeMs;
         // Check how many full windows of data we have currently retained

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

@@ -107,12 +107,15 @@ 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 */
+    // purge any samples that lack observed events within the monitored window
     protected void purgeObsoleteSamples(MetricConfig config, long now) {
         long expireAge = config.samples() * config.timeWindowMs();
         for (Sample sample : samples) {
-            if (now - sample.lastEventMs >= expireAge)
+            // samples overlapping the monitored window are kept,
+            // even if they started before it
+            if (now - sample.lastEventMs >= expireAge) {
                 sample.reset(now);
+            }
         }
     }
 
@@ -148,6 +151,7 @@ public String toString() {
                 "value=" + value +
                 ", eventCount=" + eventCount +
                 ", startTimeMs=" + startTimeMs +
+                ", lastEventMs=" + lastEventMs +
                 ", initialValue=" + initialValue +
                 ')';
         }

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

@@ -72,7 +72,8 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
 
     // 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.
+    // measurement. Start assertions from the second window. This test is to address past issue,
+    // when measurements in the end of the sample led to value spikes.
     @Test
     public void testRateIsConsistentAfterTheFirstWindow() {
         MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);