metrics() {
}
/**
- * KafkaMetrics of sensors which use SampledStat should be synchronized on the Sensor object
- * to allow concurrent reads and updates. For simplicity, all sensors are synchronized on Sensor.
+ * KafkaMetrics of sensors which use SampledStat should be synchronized on the same lock
+ * for sensor record and metric value read to allow concurrent reads and updates. For simplicity,
+ * all sensors are synchronized on this object.
+ *
+ * Sensor object is not used as a lock for reading metric value since metrics reporter is
+ * invoked while holding Sensor and Metrics locks to report addition and removal of metrics
+ * and synchronized reporters may deadlock if Sensor lock is used for reading metrics values.
+ * Note that Sensor object itself is used as a lock to protect the access to stats and metrics
+ * while recording metric values, adding and deleting sensors.
+ *
+ * Locking order (assume all MetricsReporter methods may be synchronized):
+ *
+ * - Sensor#add: Sensor -> Metrics -> MetricsReporter
+ * - Metrics#removeSensor: Sensor -> Metrics -> MetricsReporter
+ * - KafkaMetric#metricValue: MetricsReporter -> Sensor#metricLock
+ * - Sensor#record: Sensor -> Sensor#metricLock
+ *
+ *
*/
- private Object metricLock(Stat stat) {
- return this;
+ private Object metricLock() {
+ return metricLock;
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
index 6acc39d35a677..59bc84e40decf 100644
--- a/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
@@ -26,13 +26,16 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.Deque;
+import java.util.List;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.kafka.common.Metric;
@@ -54,9 +57,12 @@
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
@SuppressWarnings("deprecation")
public class MetricsTest {
+ private static final Logger log = LoggerFactory.getLogger(MetricsTest.class);
private static final double EPS = 0.000001;
private MockTime time = new MockTime();
@@ -604,8 +610,12 @@ public void testMetricInstances() {
}
}
+ /**
+ * Verifies that concurrent sensor add, remove, updates and read don't result
+ * in errors or deadlock.
+ */
@Test
- public void testConcurrentAccess() throws Exception {
+ public void testConcurrentReadUpdate() throws Exception {
final Random random = new Random();
final Deque sensors = new ConcurrentLinkedDeque<>();
metrics = new Metrics(new MockTime(10));
@@ -613,16 +623,8 @@ public void testConcurrentAccess() throws Exception {
final AtomicBoolean alive = new AtomicBoolean(true);
executorService = Executors.newSingleThreadExecutor();
- executorService.submit(new Runnable() {
- @Override
- public void run() {
- while (alive.get()) {
- for (Sensor sensor : sensors) {
- sensor.record(random.nextInt(10000));
- }
- }
- }
- });
+ executorService.submit(new ConcurrentMetricOperation(alive, "record",
+ () -> sensors.forEach(sensor -> sensor.record(random.nextInt(10000)))));
for (int i = 0; i < 10000; i++) {
if (sensors.size() > 5) {
@@ -640,6 +642,97 @@ public void run() {
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.
+ */
+ @Test
+ public void testConcurrentReadUpdateReport() throws Exception {
+
+ class LockingReporter implements MetricsReporter {
+ Map activeMetrics = new HashMap<>();
+ @Override
+ public synchronized void init(List 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 configs) {
+ }
+
+ synchronized void processMetrics() {
+ for (KafkaMetric metric : activeMetrics.values()) {
+ assertNotNull("Invalid metric value", metric.metricValue());
+ }
+ }
+ }
+
+ final LockingReporter reporter = new LockingReporter();
+ this.metrics.close();
+ this.metrics = new Metrics(config, Arrays.asList((MetricsReporter) reporter), new MockTime(10), true);
+ final Deque 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("Invalid metric value", metric.metricValue())))));
+ 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("Read failed", readFuture.isDone());
+ assertFalse("Write failed", writeFuture.isDone());
+ assertFalse("Report failed", reportFuture.isDone());
+
+ alive.set(false);
+ }
+
+ private class ConcurrentMetricOperation implements Runnable {
+ private final AtomicBoolean alive;
+ private final String opName;
+ private final Runnable op;
+ ConcurrentMetricOperation(AtomicBoolean alive, String opName, Runnable op) {
+ this.alive = alive;
+ this.opName = opName;
+ this.op = op;
+ }
+ public void run() {
+ try {
+ while (alive.get()) {
+ op.run();
+ }
+ } catch (Throwable t) {
+ log.error("Metric {} failed with exception", opName, t);
+ }
+ }
+ }
+
enum StatType {
AVG(0),
TOTAL(1),
@@ -676,7 +769,7 @@ private static class SensorCreator {
}
private Sensor createSensor(StatType statType, int index) {
- Sensor sensor = metrics.sensor("kafka.requests");
+ Sensor sensor = metrics.sensor("kafka.requests." + index);
Map tags = Collections.singletonMap("tag", "tag" + index);
switch (statType) {
case AVG: