@Test
    public void testRecordAndCheckQuotaUseMetricConfigOfEachStat() {
        final Time time = new MockTime(0, System.currentTimeMillis(), 0);
        final Metrics metrics = new Metrics(time);
        final Sensor sensor = metrics.sensor("sensor");

        final MeasurableStat stat1 = Mockito.mock(MeasurableStat.class);
        final MetricName stat1Name = metrics.metricName("stat1", "test-group");
        final MetricConfig stat1Config = new MetricConfig().quota(Quota.upperBound(5));
        sensor.add(stat1Name, stat1, stat1Config);

        final MeasurableStat stat2 = Mockito.mock(MeasurableStat.class);
        final MetricName stat2Name = metrics.metricName("stat2", "test-group");
        final MetricConfig stat2Config = new MetricConfig().quota(Quota.upperBound(10));
        sensor.add(stat2Name, stat2, stat2Config);

        sensor.record(10, 1);
        Mockito.verify(stat1).record(stat1Config, 10, 1);
        Mockito.verify(stat2).record(stat2Config, 10, 1);

        sensor.checkQuotas(2);
        Mockito.verify(stat1).measure(stat1Config, 2);
        Mockito.verify(stat2).measure(stat2Config, 2);

        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