@Test
    public void testStrictQuotaEnforcementWithRate() {
        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(11));
        final MetricName metricName = metrics.metricName("rate", "test-group");
        assertTrue(sensor.add(metricName, new Rate()));
        final KafkaMetric rateMetric = metrics.metric(metricName);

        // Recording a first value at T+0 to bring the avg rate to 3 which is already
        // above the quota.
        strictRecord(sensor, 30, time.milliseconds());
        assertEquals(3, rateMetric.measurableValue(time.milliseconds()), 0.1);

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

        // But, recording a second value is rejected because the avg rate is still equal
        // to 3 after 5s.
        assertEquals(3, rateMetric.measurableValue(time.milliseconds()), 0.1);
        assertThrows(QuotaViolationException.class, () -> strictRecord(sensor, 30, time.milliseconds()));

        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