AMBARI-20104 : Kafka Controller metrics are not being displayed. (avijayan)

ambari-metrics/ambari-metrics-kafka-sink/pom.xml

@@ -144,7 +144,7 @@ limitations under the License.
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka_2.10</artifactId>
-      <version>0.8.1.1</version>
+      <version>0.10.1.0</version>
       <exclusions>
         <exclusion>
           <groupId>com.sun.jdmk</groupId>
@@ -168,6 +168,11 @@ limitations under the License.
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+    <groupId>org.apache.kafka</groupId>
+    <artifactId>kafka-clients</artifactId>
+    <version>0.10.1.0</version>
+    </dependency>
     <dependency>
       <groupId>com.yammer.metrics</groupId>
       <artifactId>metrics-core</artifactId>

ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java

@@ -378,9 +378,12 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
       final long currentTimeMillis = System.currentTimeMillis();
       final String sanitizedName = sanitizeName(name);
 
-      cacheSanitizedTimelineMetric(currentTimeMillis, sanitizedName, "", Double.parseDouble(String.valueOf(gauge.value())));
-
-      populateMetricsList(context, MetricType.GAUGE, sanitizedName);
+      try {
+        cacheSanitizedTimelineMetric(currentTimeMillis, sanitizedName, "", Double.parseDouble(String.valueOf(gauge.value())));
+        populateMetricsList(context, MetricType.GAUGE, sanitizedName);
+      } catch (NumberFormatException ex) {
+        LOG.debug(ex.getMessage());
+      }
     }
 
     private String[] cacheKafkaMetered(long currentTimeMillis, String sanitizedName, Metered meter) {