@Test
    public void testConcurrentReadUpdateReport() {

        class LockingReporter implements MetricsReporter {
            final Map<MetricName, KafkaMetric> activeMetrics = new HashMap<>();
            @Override
            public synchronized void init(List<KafkaMetric> metrics) {
            }

            @Override
            public synchronized void metricChange(KafkaMetric metric) {
                activeMetrics.put(metric.metricName(), metric);
            }

            @Override
            public synchronized void metricRemoval(KafkaMetric metric) {
                activeMetrics.remove(metric.metricName(), metric);
            }

            @Override
            public synchronized void close() {
            }

            @Override
            public void configure(Map<String, ?> configs) {
            }

            synchronized void processMetrics() {
                for (KafkaMetric metric : activeMetrics.values()) {
                    assertNotNull(metric.metricValue(), "Invalid metric value");
                }
            }
        }

        final LockingReporter reporter = new LockingReporter();
        this.metrics.close();
        this.metrics = new Metrics(config, singletonList(reporter), new MockTime(10), true);
        final Deque<Sensor> sensors = new ConcurrentLinkedDeque<>();
        SensorCreator sensorCreator = new SensorCreator(metrics);

        final Random random = new Random();
        final AtomicBoolean alive = new AtomicBoolean(true);
        executorService = Executors.newFixedThreadPool(3);

        Future<?> writeFuture = executorService.submit(new ConcurrentMetricOperation(alive, "record",
            () -> sensors.forEach(sensor -> sensor.record(random.nextInt(10000)))));
        Future<?> readFuture = executorService.submit(new ConcurrentMetricOperation(alive, "read",
            () -> sensors.forEach(sensor -> sensor.metrics().forEach(metric ->
                assertNotNull(metric.metricValue(), "Invalid metric value")))));
        Future<?> reportFuture = executorService.submit(new ConcurrentMetricOperation(alive, "report",
                reporter::processMetrics));

        for (int i = 0; i < 10000; i++) {
            if (sensors.size() > 10) {
                Sensor sensor = random.nextBoolean() ? sensors.removeFirst() : sensors.removeLast();
                metrics.removeSensor(sensor.name());
            }
            StatType statType = StatType.forId(random.nextInt(StatType.values().length));
            sensors.add(sensorCreator.createSensor(statType, i));
        }
        assertFalse(readFuture.isDone(), "Read failed");
        assertFalse(writeFuture.isDone(), "Write failed");
        assertFalse(reportFuture.isDone(), "Report failed");

        alive.set(false);
    }
Verifies that concurrent sensor add, remove, updates and read with a metrics reporter
that synchronizes on every reporter method doesn't result in errors or deadlock.
clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java