public void maybeThrottle(double observed) {
        double currentDesiredRatePerSec = desiredRatePerSec;
        meter.mark((long) observed);
        synchronized (lock) {
            observedSoFar += observed;
            long now = time.nanoseconds();
            long elapsedNs = now - periodStartNs;
            // if we have completed an interval AND we have observed something, maybe
            // we should take a little nap
            if (elapsedNs > checkIntervalNs && observedSoFar > 0) {
                double rateInSecs = (observedSoFar * NS_PER_SEC) / elapsedNs;
                if (rateInSecs > currentDesiredRatePerSec) {
                    // solve for the amount of time to sleep to make us hit the desired rate
                    double desiredRateMs = currentDesiredRatePerSec / MS_PER_SEC;
                    long elapsedMs = TimeUnit.NANOSECONDS.toMillis(elapsedNs);
                    long sleepTime = Math.round(observedSoFar / desiredRateMs - elapsedMs);
                    if (sleepTime > 0) {
                        LOG.trace("Natural rate is {} per second but desired rate is {}, sleeping for {} ms to compensate.", rateInSecs, currentDesiredRatePerSec, sleepTime);
                        time.sleep(sleepTime);
                    }
                }
                periodStartNs = time.nanoseconds();
                observedSoFar = 0.0;
            }
        }
    }
The throttler takes a desired rate-per-second (the units of the process don't matter, it could be bytes
or a count of some other thing), and will sleep for an appropriate amount of time when maybeThrottle()
is called to attain the desired rate.

@param desiredRatePerSec  The rate we want to hit in units/sec
@param checkIntervalMs    The interval at which to check our rate
@param metricName         The name of the metric
@param units              The name of the unit
@param time               The time implementation to use
storage/src/main/java/org/apache/kafka/storage/internals/utils/Throttler.java