public long windowSize(MetricConfig config, long now) {
        // purge old samples before we compute the window size
        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
         * 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.
         *
         * 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
        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)
            totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();

        // 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
        // minimum window size to be at least 1ms.
        return Math.max(totalElapsedTimeMs, 1);
    }
The rate of the given quantity. By default this is the total observed over a set of samples from a sampled statistic
divided by the elapsed time over the sample windows. Alternative {@link SampledStat} implementations can be provided,
however, to record the rate of occurrences (e.g. the count of values measured over the time interval) or other such
values.
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java