@Test
    public void testRateWindowing() {
        // Use the default time window. Set 3 samples
        MetricConfig cfg = new MetricConfig().samples(3);
        Sensor s = metrics.sensor("test.sensor", cfg);
        MetricName rateMetricName = metrics.metricName("test.rate", "grp1");
        MetricName totalMetricName = metrics.metricName("test.total", "grp1");
        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 WindowedCount(), countRateMetricName, countTotalMetricName));
        KafkaMetric totalMetric = metrics.metrics().get(totalMetricName);
        KafkaMetric countTotalMetric = metrics.metrics().get(countTotalMetricName);

        int sum = 0;
        int count = cfg.samples() - 1;
        // Advance 1 window after every record
        for (int i = 0; i < count; i++) {
            s.record(100);
            sum += 100;
            time.sleep(cfg.timeWindowMs());
            assertEquals(sum, (Double) totalMetric.metricValue(), EPS);
        }

        // Sleep for half the window.
        time.sleep(cfg.timeWindowMs() / 2);

        // prior to any time passing, elapsedSecs = sampleWindowSize * (total samples - half of final sample)
        double elapsedSecs = MetricsUtils.convert(cfg.timeWindowMs(), TimeUnit.SECONDS) * (cfg.samples() - 0.5);

        KafkaMetric rateMetric = metrics.metrics().get(rateMetricName);
        KafkaMetric countRateMetric = metrics.metrics().get(countRateMetricName);
        assertEquals(sum / elapsedSecs, (Double) rateMetric.metricValue(), EPS, "Rate(0...2) = 2.666");
        assertEquals(count / elapsedSecs, (Double) countRateMetric.metricValue(), EPS, "Count rate(0...2) = 0.02666");
        assertEquals(elapsedSecs,
                MetricsUtils.convert(((Rate) rateMetric.measurable()).windowSize(cfg, time.milliseconds()), TimeUnit.SECONDS), EPS, "Elapsed Time = 75 seconds");
        assertEquals(sum, (Double) totalMetric.metricValue(), EPS);
        assertEquals(count, (Double) countTotalMetric.metricValue(), EPS);

        // Verify that rates are expired, but total is cumulative
        time.sleep(cfg.timeWindowMs() * cfg.samples());
        assertEquals(0, (Double) rateMetric.metricValue(), EPS);
        assertEquals(0, (Double) countRateMetric.metricValue(), EPS);
        assertEquals(sum, (Double) totalMetric.metricValue(), EPS);
        assertEquals(count, (Double) countTotalMetric.metricValue(), EPS);
    }
Some implementations of SampledStat make sense to return the initial value
when there are no values set
clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java