kamalcph commented on code in PR #13969:
URL: https://github.com/apache/kafka/pull/13969#discussion_r1257268739
##########
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:
Will the test throw exceptions consistently? What if
`sensor.checkQuotasNonThreadSafe(3)` ran before we are adding a new metric?
##########
clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java:
##########
@@ -250,7 +250,16 @@ public void checkQuotas() {
checkQuotas(time.milliseconds());
}
- public void checkQuotas(long timeMs) {
+ public synchronized void checkQuotas(long timeMs) {
+ checkQuotasInternal(timeMs);
+ }
+
+ // visible for testing
+ protected void checkQuotasNonThreadSafe(long timeMs) {
Review Comment:
Can we make it package-private instead of protected?
--
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
