@Test
    public void testStrictQuotaEnforcementWithTokenBucket() {
        final Time time = new MockTime(0, System.currentTimeMillis(), 0);
        final Metrics metrics = new Metrics(time);
        final Sensor sensor = metrics.sensor("sensor", new MetricConfig()
            .quota(Quota.upperBound(2))
            .timeWindow(1, TimeUnit.SECONDS)
            .samples(10));
        final MetricName metricName = metrics.metricName("credits", "test-group");
        assertTrue(sensor.add(metricName, new TokenBucket()));
        final KafkaMetric tkMetric = metrics.metric(metricName);

        // Recording a first value at T+0 to bring the remaining credits below zero
        strictRecord(sensor, 30, time.milliseconds());
        assertEquals(-10, tkMetric.measurableValue(time.milliseconds()), 0.1);

        // Theoretically, we should wait 5s to bring back the avg rate to the define quota:
        // 10 / 2 = 5s
        time.sleep(5000);

        // Unlike the default rate based on a windowed sum, it works as expected.
        assertEquals(0, tkMetric.measurableValue(time.milliseconds()), 0.1);
        strictRecord(sensor, 30, time.milliseconds());
        assertEquals(-30, tkMetric.measurableValue(time.milliseconds()), 0.1);

        metrics.close();
    }
The Sensor#checkQuotas should be thread-safe since the method may be used by many ReplicaFetcherThreads.
clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java