[KAFKA-14119] Sensor in metrics has potential thread safety issues

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Not A Bug
• Component/s: metrics
• Assignee: Unassigned
• Reporter: Eric Wu
• Created: 27/Jul/22 19:16
• Updated: 27/Jul/22 20:10
• Resolved: 27/Jul/22 20:10

Description

There are potential issues of a `Sensor` not being protected from race conditions when it records a value.

It can be reproduced with a unit test, e.g., in `SensorTest`:

@Test
public void testSensorRecordThreadSafety() {
    Time time = new MockTime(0, System.currentTimeMillis(), 0);
    Metrics metrics = new Metrics(time);
    Sensor sensor = metrics.sensor("sensor");
    MetricName metric = new MetricName("test", "test", "test", Collections.emptyMap());
    sensor.add(metric, new Value());

    int totalRequests = 10;
    AtomicInteger count = new AtomicInteger();
    Executor threadPool = Executors.newFixedThreadPool(totalRequests);
    CompletableFuture[] futures = new CompletableFuture[totalRequests];
    for (int i = 0; i < totalRequests; i++) {
        futures[i] = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(10); // to make it easier to repro
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            sensor.record(count.addAndGet(1));
        }, threadPool);
    }
    CompletableFuture.allOf(futures).join();
    
    assertEquals(1, sensor.metrics().size());
    double value = (double) sensor.metrics().get(0).metricValue();
    assertEquals(totalRequests, value);
}

It needs some tweaks to make the fields visible in the test. Given a few runs, the test should fail which demonstrates the thread safety issue.