kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject [kafka] branch trunk updated: MINOR: make Sensor#add idempotent (#4853)
Date Thu, 12 Apr 2018 03:50:16 GMT
This is an automated email from the ASF dual-hosted git repository.

guozhang pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new cc43e77  MINOR: make Sensor#add idempotent (#4853)
cc43e77 is described below

commit cc43e77bbbfad71883011186de55603c936cbcd1
Author: John Roesler <vvcephei@users.noreply.github.com>
AuthorDate: Wed Apr 11 22:50:10 2018 -0500

    MINOR: make Sensor#add idempotent (#4853)
    
    This change makes adding a metric to a sensor idempotent.
    That is, if the metric is already added to the sensor, the method
    returns with success.
    
    The current behavior is that any attempt to register a second metric
    with the same name is an error.
    
    Testing strategy: There is a new unit test covering this behavior
    
    Reviewers: Guozhang Wang <guozhang@confluent.io>, Bill Bejeck <bill@confluent.io>
---
 .../org/apache/kafka/common/metrics/Sensor.java    | 53 +++++++++++++---------
 .../apache/kafka/common/metrics/SensorTest.java    | 48 ++++++++++++++++----
 .../kafka/connect/runtime/ConnectMetricsTest.java  | 25 +++++-----
 3 files changed, 83 insertions(+), 43 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
index 06c8c7f..22f273d 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
@@ -24,8 +24,11 @@ import org.apache.kafka.common.utils.Utils;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -40,7 +43,7 @@ public final class Sensor {
     private final String name;
     private final Sensor[] parents;
     private final List<Stat> stats;
-    private final List<KafkaMetric> metrics;
+    private final Map<MetricName, KafkaMetric> metrics;
     private final MetricConfig config;
     private final Time time;
     private volatile long lastRecordTime;
@@ -103,7 +106,7 @@ public final class Sensor {
         this.registry = registry;
         this.name = Utils.notNull(name);
         this.parents = parents == null ? new Sensor[0] : parents;
-        this.metrics = new ArrayList<>();
+        this.metrics = new LinkedHashMap<>();
         this.stats = new ArrayList<>();
         this.config = config;
         this.time = time;
@@ -190,7 +193,7 @@ public final class Sensor {
     }
 
     public void checkQuotas(long timeMs) {
-        for (KafkaMetric metric : this.metrics) {
+        for (KafkaMetric metric : this.metrics.values()) {
             MetricConfig config = metric.config();
             if (config != null) {
                 Quota quota = config.quota();
@@ -228,9 +231,11 @@ public final class Sensor {
         this.stats.add(Utils.notNull(stat));
         Object lock = new Object();
         for (NamedMeasurable m : stat.stats()) {
-            KafkaMetric metric = new KafkaMetric(lock, m.name(), m.stat(), config == null
? this.config : config, time);
-            this.registry.registerMetric(metric);
-            this.metrics.add(metric);
+            final KafkaMetric metric = new KafkaMetric(lock, m.name(), m.stat(), config ==
null ? this.config : config, time);
+            if (!metrics.containsKey(metric.metricName())) {
+                registry.registerMetric(metric);
+                metrics.put(metric.metricName(), metric);
+            }
         }
         return true;
     }
@@ -247,24 +252,30 @@ public final class Sensor {
 
     /**
      * Register a metric with this sensor
+     *
      * @param metricName The name of the metric
-     * @param stat The statistic to keep
-     * @param config A special configuration for this metric. If null use the sensor default
configuration.
+     * @param stat       The statistic to keep
+     * @param config     A special configuration for this metric. If null use the sensor
default configuration.
      * @return true if metric is added to sensor, false if sensor is expired
      */
-    public synchronized boolean add(MetricName metricName, MeasurableStat stat, MetricConfig
config) {
-        if (hasExpired())
+    public synchronized boolean add(final MetricName metricName, final MeasurableStat stat,
final MetricConfig config) {
+        if (hasExpired()) {
             return false;
-
-        KafkaMetric metric = new KafkaMetric(new Object(),
-                                             Utils.notNull(metricName),
-                                             Utils.notNull(stat),
-                                             config == null ? this.config : config,
-                                             time);
-        this.registry.registerMetric(metric);
-        this.metrics.add(metric);
-        this.stats.add(stat);
-        return true;
+        } else if (metrics.containsKey(metricName)) {
+            return true;
+        } else {
+            final KafkaMetric metric = new KafkaMetric(
+                new Object(),
+                Utils.notNull(metricName),
+                Utils.notNull(stat),
+                config == null ? this.config : config,
+                time
+            );
+            registry.registerMetric(metric);
+            metrics.put(metric.metricName(), metric);
+            stats.add(stat);
+            return true;
+        }
     }
 
     /**
@@ -276,6 +287,6 @@ public final class Sensor {
     }
 
     synchronized List<KafkaMetric> metrics() {
-        return Collections.unmodifiableList(this.metrics);
+        return Collections.unmodifiableList(new LinkedList<>(this.metrics.values()));
     }
 }
diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
index 3f7551e..a1c0814 100644
--- a/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
@@ -16,23 +16,25 @@
  */
 package org.apache.kafka.common.metrics;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.stats.Avg;
 import org.apache.kafka.common.metrics.stats.Meter;
+import org.apache.kafka.common.metrics.stats.Sum;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.kafka.common.utils.Time;
 import org.junit.Test;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
 public class SensorTest {
     @Test
     public void testRecordLevelEnum() {
@@ -94,4 +96,32 @@ public class SensorTest {
 
         metrics.close();
     }
+
+    @Test
+    public void testIdempotentAdd() {
+        final Metrics metrics = new Metrics();
+        final Sensor sensor = metrics.sensor("sensor");
+
+        assertTrue(sensor.add(metrics.metricName("test-metric", "test-group"), new Avg()));
+
+        // adding the same metric to the same sensor is a no-op
+        assertTrue(sensor.add(metrics.metricName("test-metric", "test-group"), new Avg()));
+
+
+        // but adding the same metric to a DIFFERENT sensor is an error
+        final Sensor anotherSensor = metrics.sensor("another-sensor");
+        try {
+            anotherSensor.add(metrics.metricName("test-metric", "test-group"), new Avg());
+            fail("should have thrown");
+        } catch (final IllegalArgumentException ignored) {
+            // pass
+        }
+
+        // note that adding a different metric with the same name is also a no-op
+        assertTrue(sensor.add(metrics.metricName("test-metric", "test-group"), new Sum()));
+
+        // so after all this, we still just have the original metric registered
+        assertEquals(1, sensor.metrics().size());
+        assertEquals(org.apache.kafka.common.metrics.stats.Avg.class, sensor.metrics().get(0).measurable().getClass());
+    }
 }
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ConnectMetricsTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ConnectMetricsTest.java
index 60bd863..25395f9 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ConnectMetricsTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ConnectMetricsTest.java
@@ -20,11 +20,10 @@ import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.Avg;
 import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup;
 import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroupId;
-import org.apache.kafka.common.utils.MockTime;
 import org.junit.After;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -145,21 +144,21 @@ public class ConnectMetricsTest {
 
     @Test
     public void testRecreateWithClose() {
-        int numMetrics = addToGroup(metrics, false);
-        int numMetricsInRecreatedGroup = addToGroup(metrics, true);
-        Assert.assertEquals(numMetrics, numMetricsInRecreatedGroup);
+        final Sensor originalSensor = addToGroup(metrics, false);
+        final Sensor recreatedSensor = addToGroup(metrics, true);
+        // because we closed the metricGroup, we get a brand-new sensor
+        assertNotSame(originalSensor, recreatedSensor);
     }
 
-    @Test(expected = IllegalArgumentException.class)
+    @Test
     public void testRecreateWithoutClose() {
-        int numMetrics = addToGroup(metrics, false);
-        int numMetricsInRecreatedGroup = addToGroup(metrics, false);
-        // we should never get here
-        throw new RuntimeException("Created " + numMetricsInRecreatedGroup
-                + " metrics in recreated group. Original=" + numMetrics);
+        final Sensor originalSensor = addToGroup(metrics, false);
+        final Sensor recreatedSensor = addToGroup(metrics, false);
+        // since we didn't close the group, the second addToGroup is idempotent
+        assertSame(originalSensor, recreatedSensor);
     }
 
-    private int addToGroup(ConnectMetrics connectMetrics, boolean shouldClose) {
+    private Sensor addToGroup(ConnectMetrics connectMetrics, boolean shouldClose) {
         ConnectMetricsRegistry registry = connectMetrics.registry();
         ConnectMetrics.MetricGroup metricGroup = connectMetrics.group(registry.taskGroupName(),
                 registry.connectorTagName(), "conn_name");
@@ -172,7 +171,7 @@ public class ConnectMetricsTest {
         sensor.add(metricName("x1"), new Max());
         sensor.add(metricName("y2"), new Avg());
 
-        return metricGroup.metrics().metrics().size();
+        return sensor;
     }
 
     static MetricName metricName(String name) {

-- 
To stop receiving notification emails like this one, please contact
guozhang@apache.org.

Mime
View raw message