Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java (Java)

Code before:
    public Rate(TimeUnit unit) {
        this(unit, new WindowedSum());
    }

    public Rate(SampledStat stat) {
        this(TimeUnit.SECONDS, stat);
    }

    public Rate(TimeUnit unit, SampledStat stat) {
        this(unit, stat, -1);
    }

    public Rate(TimeUnit unit, SampledStat stat, long window) {
        this.unit = unit;
        this.stat = stat;
        this.window = window;
    }

    public String unitName() {
        return unit.name().substring(0, unit.name().length() - 2).toLowerCase(Locale.ROOT);
    }

    @Override
    public void record(MetricConfig config, double value, long timeMs) {
        if (window != -1)
            config = new MetricConfig(config).timeWindow(window, unit);
        this.stat.record(config, value, timeMs);
    }

    @Override
    public double measure(MetricConfig config, long now) {
        if (window != -1)
            config = new MetricConfig(config).timeWindow(window, unit);

        double value = stat.measure(config, now);
        return value / convert(windowSize(config, now), unit);
    }

    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,

Review comment (suggestion):
Not appropriate to create a MetricConfig on every `record` execution. Rather we can keep the `MetricConfig` as instance variable.

Code after:
    public Rate(TimeUnit unit) {
        this(unit, new WindowedSum());
    }

    public Rate(SampledStat stat) {
        this(TimeUnit.SECONDS, stat);
    }

    public Rate(TimeUnit unit, SampledStat stat) {
        this(unit, stat, -1);
    }

    public Rate(TimeUnit unit, SampledStat stat, long window) {
        this.unit = unit;
        this.stat = stat;
        if (window > 0) {
            this.timeWindowMs = TimeUnit.MILLISECONDS.convert(window, unit);
            this.stat.withTimeWindow(window, unit);
        } else  {
            this.timeWindowMs = -1;
        }
    }

    public String unitName() {
        return unit.name().substring(0, unit.name().length() - 2).toLowerCase(Locale.ROOT);
    }

    @Override
    public void record(MetricConfig config, double value, long timeMs) {
        this.stat.record(config, value, timeMs);
    }

    @Override
    public double measure(MetricConfig config, long now) {
        double value = stat.measure(config, now);
        return value / convert(windowSize(config, now), unit);
    }

    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,