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;
            }
        }
    }
@Test
    public void testThrottleDesiredRate() {
        long throttleCheckIntervalMs = 100L;
        double desiredCountPerSec = 1000.0;
        double desiredCountPerInterval = desiredCountPerSec * throttleCheckIntervalMs / 1000.0;

        Time mockTime = new MockTime();
        Throttler throttler = new Throttler(desiredCountPerSec,
                                            throttleCheckIntervalMs,
                                            "throttler",
                                            "entries",
                                            mockTime);

        // Observe desiredCountPerInterval at t1
        long t1 = mockTime.milliseconds();
        throttler.maybeThrottle(desiredCountPerInterval);
        assertEquals(t1, mockTime.milliseconds());

        // Observe desiredCountPerInterval at t1 + throttleCheckIntervalMs + 1,
        mockTime.sleep(throttleCheckIntervalMs + 1);
        throttler.maybeThrottle(desiredCountPerInterval);
        long t2 = mockTime.milliseconds();
        assertTrue(t2 >= t1 + 2 * throttleCheckIntervalMs);

        // Observe desiredCountPerInterval at t2
        throttler.maybeThrottle(desiredCountPerInterval);
        assertEquals(t2, mockTime.milliseconds());

        // Observe desiredCountPerInterval at t2 + throttleCheckIntervalMs + 1
        mockTime.sleep(throttleCheckIntervalMs + 1);
        throttler.maybeThrottle(desiredCountPerInterval);
        long t3 = mockTime.milliseconds();
        assertTrue(t3 >= t2 + 2 * throttleCheckIntervalMs);

        long elapsedTimeMs = t3 - t1;
        double actualCountPerSec = 4 * desiredCountPerInterval * 1000 / elapsedTimeMs;
        assertTrue(actualCountPerSec <= desiredCountPerSec);
    }