diff --git a/pom.xml b/pom.xml
index 68cf22a..1ef8bf8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -13,8 +13,9 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <maven.compiler.source>1.7</maven.compiler.source>
-        <maven.compiler.target>1.7</maven.compiler.target>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
+        <dropwizard.metrics.version>4.1.12.1</dropwizard.metrics.version>
     </properties>
 
     <developers>
@@ -79,7 +80,7 @@
         <dependency>
             <groupId>io.dropwizard.metrics</groupId>
             <artifactId>metrics-core</artifactId>
-            <version>3.1.2</version>
+            <version>${dropwizard.metrics.version}</version>
         </dependency>
 
         <!-- Optional; the user is expected to specify this dependency
@@ -87,7 +88,7 @@
         <dependency>
             <groupId>io.dropwizard.metrics</groupId>
             <artifactId>metrics-graphite</artifactId>
-            <version>3.1.2</version>
+            <version>${dropwizard.metrics.version}</version>
             <scope>provided</scope>
         </dependency>
 
@@ -96,7 +97,7 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
-            <version>0.9.0.1</version>
+            <version>3.2.1</version>
             <scope>provided</scope>
         </dependency>
 
diff --git a/src/main/java/com/simple/metrics/kafka/DropwizardReporter.java b/src/main/java/com/simple/metrics/kafka/DropwizardReporter.java
index f9fed63..22a516b 100644
--- a/src/main/java/com/simple/metrics/kafka/DropwizardReporter.java
+++ b/src/main/java/com/simple/metrics/kafka/DropwizardReporter.java
@@ -21,7 +21,7 @@ public class DropwizardReporter implements MetricsReporter {
 
     protected MetricRegistry registry;
     protected DropwizardReporterConfig config;
-    private Set<String> metricNames = new HashSet<>();
+    private final Set<String> metricNames = new HashSet<>();
 
     @Override
     public void configure(Map<String, ?> configs) {
@@ -45,10 +45,12 @@ public void metricChange(final KafkaMetric kafkaMetric) {
         LOGGER.debug("Processing a metric change for {}", kafkaMetric.metricName().toString());
         String name = dropwizardMetricName(kafkaMetric);
 
-        Gauge<Double> gauge = new Gauge<Double>() {
-            @Override
-            public Double getValue() {
-                return kafkaMetric.value();
+        Gauge<Double> gauge = () -> {
+            if (kafkaMetric.metricValue() instanceof Number) {
+                return ((Number) kafkaMetric.metricValue()).doubleValue();
+            } else {
+                // Null values are not reported by GraphiteReporter
+                return null;
             }
         };
         LOGGER.debug("Registering {}", name);
@@ -70,25 +72,21 @@ public void metricRemoval(KafkaMetric kafkaMetric) {
 
     @Override
     public void close() {
-        for (String name: metricNames)
+        for (String name : metricNames)
             registry.remove(name);
     }
 
     private static String dropwizardMetricName(KafkaMetric kafkaMetric) {
         MetricName name = kafkaMetric.metricName();
 
-        List<String> nameParts = new ArrayList<String>(2);
+        List<String> nameParts = new ArrayList<>(2);
         nameParts.add(name.group());
         nameParts.addAll(name.tags().values());
         nameParts.add(name.name());
 
-        StringBuilder builder = new StringBuilder();
-        for (String namePart : nameParts) {
-            builder.append(namePart);
-            builder.append(".");
-        }
-        builder.setLength(builder.length() - 1);  // Remove the trailing dot.
-        String processedName = builder.toString().replace(' ', '_').replace("\\.", "_");
+        String processedName = String.join(".", nameParts)
+                .replace(' ', '_')
+                .replace("\\.", "_");
 
         return MetricRegistry.name(METRIC_PREFIX, processedName);
     }
diff --git a/src/test/java/com/simple/metrics/kafka/DropwizardReporterTest.java b/src/test/java/com/simple/metrics/kafka/DropwizardReporterTest.java
index 1a0e761..f7343c1 100644
--- a/src/test/java/com/simple/metrics/kafka/DropwizardReporterTest.java
+++ b/src/test/java/com/simple/metrics/kafka/DropwizardReporterTest.java
@@ -3,6 +3,7 @@
 import com.codahale.metrics.Gauge;
 import com.codahale.metrics.SharedMetricRegistries;
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.Avg;
@@ -13,23 +14,74 @@
 import java.util.Map;
 
 public class DropwizardReporterTest {
+    @SuppressWarnings("rawtypes")
     @Test
-    public void testMetricChange() throws Exception {
+    public void testMetricChange() {
         Metrics metrics = new Metrics();
         DropwizardReporter reporter = new DropwizardReporter();
-        reporter.configure(new HashMap<String, Object>());
+        reporter.configure(new HashMap<>());
         metrics.addReporter(reporter);
         Sensor sensor = metrics.sensor("kafka.requests");
-        sensor.add(new MetricName("pack.bean1.avg", "grp1"), new Avg());
+        sensor.add(new MetricName("pack.bean1.avg", "grp1", "", new HashMap<>()), new Avg());
 
         Map<String, Gauge> gauges = SharedMetricRegistries.getOrCreate("default").getGauges();
         String expectedName = "org.apache.kafka.common.metrics.grp1.pack.bean1.avg";
-        Assert.assertEquals(1, gauges.size());
-        Assert.assertEquals(expectedName, gauges.keySet().toArray()[0]);
+        Assert.assertTrue(gauges.containsKey(expectedName));
 
         sensor.record(2.1);
         sensor.record(2.2);
         sensor.record(2.6);
-        Assert.assertEquals(2.3, (Double)gauges.get(expectedName).getValue(), 0.001);
+        Assert.assertEquals(2.3, (Double) gauges.get(expectedName).getValue(), 0.001);
+    }
+
+    @Test
+    public void testNonNumericMetricYieldsNullValue() {
+        Metrics metrics = new Metrics();
+        DropwizardReporter reporter = new DropwizardReporter();
+        reporter.configure(new HashMap<>());
+        metrics.addReporter(reporter);
+        metrics.addMetric(new MetricName("metric", "non-numeric", "", new HashMap<>()), new ConstantGauge<>("non-numeric"));
+        String expectedName = "org.apache.kafka.common.metrics.non-numeric.metric";
+        Assert.assertTrue(reporter.registry.getGauges().containsKey(expectedName));
+        Assert.assertNull(reporter.registry.getGauges().get(expectedName).getValue());
+    }
+
+    @Test
+    public void testNumericMetricYieldsActualValue() {
+        Metrics metrics = new Metrics();
+        DropwizardReporter reporter = new DropwizardReporter();
+        reporter.configure(new HashMap<>());
+        metrics.addReporter(reporter);
+        metrics.addMetric(new MetricName("metric", "numeric", "", new HashMap<>()),
+                new ConstantGauge<>(12));
+        String expectedName = "org.apache.kafka.common.metrics.numeric.metric";
+        Assert.assertTrue(reporter.registry.getGauges().containsKey(expectedName));
+        Assert.assertEquals((Double) reporter.registry.getGauges().get(expectedName).getValue(), 12.0, 0.001);
+    }
+
+    @Test
+    public void testNullMetricYieldsNull() {
+        Metrics metrics = new Metrics();
+        DropwizardReporter reporter = new DropwizardReporter();
+        reporter.configure(new HashMap<>());
+        metrics.addReporter(reporter);
+        metrics.addMetric(new MetricName("metric", "numeric", "", new HashMap<>()),
+                new ConstantGauge<Double>(null));
+        String expectedName = "org.apache.kafka.common.metrics.numeric.metric";
+        Assert.assertTrue(reporter.registry.getGauges().containsKey(expectedName));
+        Assert.assertEquals((Double) reporter.registry.getGauges().get(expectedName).getValue(), 12.0, 0.001);
+    }
+
+    private static class ConstantGauge<T> implements org.apache.kafka.common.metrics.Gauge<T> {
+        private final T value;
+
+        public ConstantGauge(T value) {
+            this.value = value;
+        }
+
+        @Override
+        public T value(MetricConfig metricConfig, long l) {
+            return value;
+        }
     }
 }