extracted method startReporters

mesos · Jul 29, 2015 · def0611 · def0611
1 parent 4ac4f67
commit def0611
Showing 1 changed file with 10 additions and 2 deletions.
diff --git a/src/scala/ly/stealth/mesos/kafka/BrokerServer.scala b/src/scala/ly/stealth/mesos/kafka/BrokerServer.scala
@@ -42,6 +42,7 @@ class KafkaServer extends BrokerServer {
 
     BrokerServer.Distro.configureLog4j(broker.log4jOptions)
     val options = broker.options(defaults)
+    BrokerServer.Distro.startReporters(options)
 
     logger.info("Starting KafkaServer")
     server = BrokerServer.Distro.newServer(options)
@@ -80,12 +81,19 @@ object BrokerServer {
       val config: Object = configClass.getConstructor(classOf[Properties]).newInstance(props).asInstanceOf[Object]
       val server: Object = serverClass.getConstructor(configClass).newInstance(config).asInstanceOf[Object]
 
+      server
+    }
+
+    def startReporters(options: util.Map[String, String]): Object = {
+      val configClass = loader.loadClass("kafka.server.KafkaConfig")
+
+      val props: Properties = this.props(options, "server.properties")
+      val config: Object = configClass.getConstructor(classOf[Properties]).newInstance(props).asInstanceOf[Object]
+
       val metricsReporter = loader.loadClass("kafka.metrics.KafkaMetricsReporter$").getField("MODULE$").get(null)
       val metricsReporterClass = metricsReporter.getClass
       val verifiableProps = config.getClass.getMethod("props").invoke(config)
       metricsReporterClass.getMethod("startReporters", verifiableProps.getClass).invoke(metricsReporter, verifiableProps)
-
-      server
     }
 
     def configureLog4j(options: util.Map[String, String]): Unit = {