DL1231 commented on code in PR #13969:
URL: https://github.com/apache/kafka/pull/13969#discussion_r1263954988
##########
clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java:
##########
@@ -366,4 +368,60 @@ public void
testUpdatingMetricConfigIsReflectedInTheSensor() {
metrics.close();
}
+
+ @Test
+ public void testConcurrentReadWriteAccessForMetrics() throws
ExecutionException, InterruptedException {
+ final Metrics metrics = new Metrics(new
MetricConfig().quota(Quota.upperBound(Double.MAX_VALUE))
+ .timeWindow(1, TimeUnit.MILLISECONDS)
+ .samples(100));
+ final Sensor sensor = metrics.sensor("sensor");
+
+ final int metricCount = 10000;
+ // Add a large number of metrics to increase the execution time of
checkQuotas
+ for (int i = 0; i < metricCount; i++) {
+ sensor.add(metrics.metricName("test-metric" + i, "test-group" +
i), new Rate());
+ }
+ sensor.record(10, 1, false);
+ CountDownLatch latch = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ // Use non-thread-safe methods for concurrent read and write of
metrics.
+ Future<Throwable> worker = executor.submit(() -> {
+ try {
+ sensor.checkQuotasNonThreadSafe(3);
+ return null;
+ } catch (Throwable e) {
+ return e;
+ }
+ });
+ executor.submit(() -> {
+ sensor.add(metrics.metricName("test-metric-non-thread-safe",
"test-group-non-thread-safe"), new Rate());
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS), "If this failure happen
frequently, we can try to increase the wait time");
+ assertTrue(worker.isDone());
+ assertNotNull(worker.get());
+ assertEquals(ConcurrentModificationException.class,
worker.get().getClass());
Review Comment:
@divijvaidya thank you very much for your explanation. However, I still have
a question. Both 'add' and 'checkQuotas' are synchronized, so how can they be
accessed simultaneously?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
