AMBARI-18170 : Kafka Metrics do not show up in AMS HA enabled cluster. (avijayan)

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

@@ -61,7 +61,7 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
 
   private static final String TIMELINE_METRICS_SEND_INTERVAL_PROPERTY = "kafka.timeline.metrics.sendInterval";
   private static final String TIMELINE_METRICS_MAX_ROW_CACHE_SIZE_PROPERTY = "kafka.timeline.metrics.maxRowCacheSize";
-  private static final String TIMELINE_HOST_PROPERTY = "kafka.timeline.metrics.host";
+  private static final String TIMELINE_HOSTS_PROPERTY = "kafka.timeline.metrics.hosts";
   private static final String TIMELINE_PORT_PROPERTY = "kafka.timeline.metrics.port";
   private static final String TIMELINE_PROTOCOL_PROPERTY = "kafka.timeline.metrics.protocol";
   private static final String TIMELINE_REPORTER_ENABLED_PROPERTY = "kafka.timeline.metrics.reporter.enabled";
@@ -74,7 +74,6 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
   private volatile boolean initialized = false;
   private boolean running = false;
   private final Object lock = new Object();
-  private String collectorUri;
   private String hostname;
   private String metricCollectorPort;
   private String collectors;
@@ -147,17 +146,12 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
         int maxRowCacheSize = props.getInt(TIMELINE_METRICS_MAX_ROW_CACHE_SIZE_PROPERTY, MAX_RECS_PER_NAME_DEFAULT);
 
         zookeeperQuorum = props.getString("zookeeper.connect");
-        collectors = props.getString(TIMELINE_HOST_PROPERTY, TIMELINE_DEFAULT_HOST);
-        metricCollectorProtocol = props.getString(TIMELINE_PROTOCOL_PROPERTY, TIMELINE_DEFAULT_PROTOCOL);
-
-        String metricCollectorHost = props.getString(TIMELINE_HOST_PROPERTY, TIMELINE_DEFAULT_HOST);
         metricCollectorPort = props.getString(TIMELINE_PORT_PROPERTY, TIMELINE_DEFAULT_PORT);
+        collectors = props.getString(TIMELINE_HOSTS_PROPERTY, TIMELINE_DEFAULT_HOST + ":" + metricCollectorPort);
+        metricCollectorProtocol = props.getString(TIMELINE_PROTOCOL_PROPERTY, TIMELINE_DEFAULT_PROTOCOL);
 
         setMetricsCache(new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval));
 
-        collectorUri = constructTimelineMetricUri(metricCollectorProtocol,
-          metricCollectorHost, metricCollectorPort);
-
         if (metricCollectorProtocol.contains("https")) {
           String trustStorePath = props.getString(SSL_KEYSTORE_PATH_PROPERTY).trim();
           String trustStoreType = props.getString(SSL_KEYSTORE_TYPE_PROPERTY).trim();
@@ -181,7 +175,6 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
           startReporter(metricsConfig.pollingIntervalSecs());
         }
         if (LOG.isDebugEnabled()) {
-          LOG.debug("CollectorUri = " + collectorUri);
           LOG.debug("MetricsSendInterval = " + metricsSendInterval);
           LOG.debug("MaxRowCacheSize = " + maxRowCacheSize);
           LOG.debug("Excluded metrics prefixes = " + excludedMetricsStr);

ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterTest.java

@@ -79,7 +79,7 @@ public class KafkaTimelineMetricsReporterTest {
     properties.setProperty("zookeeper.connect", "localhost:2181");
     properties.setProperty("kafka.timeline.metrics.sendInterval", "5900");
     properties.setProperty("kafka.timeline.metrics.maxRowCacheSize", "10000");
-    properties.setProperty("kafka.timeline.metrics.host", "localhost");
+    properties.setProperty("kafka.timeline.metrics.hosts", "localhost:6188");
     properties.setProperty("kafka.timeline.metrics.port", "6188");
     properties.setProperty("kafka.timeline.metrics.reporter.enabled", "true");
     properties.setProperty("external.kafka.metrics.exclude.prefix", "a.b.c");

ambari-server/src/main/resources/common-services/KAFKA/0.8.1/configuration/kafka-broker.xml

@@ -357,8 +357,8 @@
     <on-ambari-upgrade add="true"/>
   </property>
   <property>
-    <name>kafka.timeline.metrics.host</name>
-    <value>{{metric_collector_host}}</value>
+    <name>kafka.timeline.metrics.hosts</name>
+    <value>{{metric_collector_hosts}}</value>
     <description>Timeline host</description>
     <on-ambari-upgrade add="true"/>
   </property>

ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/kafka.py

@@ -99,7 +99,7 @@ def kafka(upgrade_type=None):
       kafka_server_config['host.name'] = params.hostname
 
     if params.has_metric_collector:
-      kafka_server_config['kafka.timeline.metrics.host'] = params.metric_collector_host
+      kafka_server_config['kafka.timeline.metrics.hosts'] = params.metric_collector_hosts
       kafka_server_config['kafka.timeline.metrics.port'] = params.metric_collector_port
       kafka_server_config['kafka.timeline.metrics.protocol'] = params.metric_collector_protocol
       kafka_server_config['kafka.timeline.metrics.truststore.path'] = params.metric_truststore_path

ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/params.py

@@ -144,6 +144,14 @@ if has_metric_collector:
   else:
     metric_collector_protocol = 'http'
   pass
+
+  # Collector hosts
+  metric_collector_hosts = ""
+  if ams_collector_hosts:
+    for host in ams_collector_hosts:
+      metric_collector_hosts += host + ':' + metric_collector_port + ','
+    metric_collector_hosts = metric_collector_hosts[:-1]
+
 # Security-related params
 security_enabled = config['configurations']['cluster-env']['security_enabled']
 kafka_kerberos_enabled = (('security.inter.broker.protocol' in config['configurations']['kafka-broker']) and

ambari-server/src/test/python/stacks/2.2/configs/ranger-admin-upgrade.json

@@ -455,7 +455,7 @@
             "replica.high.watermark.checkpoint.interval.ms": "5000", 
             "zookeeper.connect": "c6407.ambari.apache.org:2181,c6408.ambari.apache.org:2181,c6409.ambari.apache.org:2181", 
             "controlled.shutdown.retry.backoff.ms": "5000", 
-            "kafka.timeline.metrics.host": "{{metric_collector_host}}", 
+            "kafka.timeline.metrics.host": "{{metric_collector_hosts}}",
             "kafka.timeline.metrics.reporter.sendInterval": "5900", 
             "num.partitions": "1", 
             "log.flush.interval.messages": "10000", 

ambari-server/src/test/python/stacks/2.2/configs/ranger-usersync-upgrade.json

@@ -450,7 +450,7 @@
             "replica.high.watermark.checkpoint.interval.ms": "5000", 
             "zookeeper.connect": "c6407.ambari.apache.org:2181,c6408.ambari.apache.org:2181,c6409.ambari.apache.org:2181", 
             "controlled.shutdown.retry.backoff.ms": "5000", 
-            "kafka.timeline.metrics.host": "{{metric_collector_host}}", 
+            "kafka.timeline.metrics.host": "{{metric_collector_hosts}}",
             "kafka.timeline.metrics.reporter.sendInterval": "5900", 
             "num.partitions": "1", 
             "log.flush.interval.messages": "10000", 

ambari-web/app/assets/data/stacks/HDP-2.2/configurations.json

@@ -8800,13 +8800,13 @@
           }
         },
         {
-          "href" : "http://c6401:8080/api/v1/stacks/HDP/versions/2.2/services/KAFKA/configurations/kafka.timeline.metrics.host",
+          "href" : "http://c6401:8080/api/v1/stacks/HDP/versions/2.2/services/KAFKA/configurations/kafka.timeline.metrics.hosts",
           "StackConfigurations" : {
             "final" : "false",
             "property_description" : "Timeline host",
-            "property_name" : "kafka.timeline.metrics.host",
+            "property_name" : "kafka.timeline.metrics.hosts",
             "property_type" : [ ],
-            "property_value" : "{{metric_collector_host}}",
+            "property_value" : "{{metric_collector_hosts}}",
             "service_name" : "KAFKA",
             "stack_name" : "HDP",
             "stack_version" : "2.2",