KAFKA-2671: Enable starting Kafka server with a Properties object

Author: Ashish Singh <[email protected]>

Reviewers: Eno Thereska, Gwen Shapira

Closes apache#330 from SinghAsDev/KAFKA-2671

Author: Ashish Singh

Diff for: core/src/main/scala/kafka/Kafka.scala

@@ -19,13 +19,13 @@ package kafka
 
 import java.util.Properties
 
-import scala.collection.JavaConversions._
 import joptsimple.OptionParser
-import metrics.KafkaMetricsReporter
-import server.{KafkaConfig, KafkaServerStartable, KafkaServer}
-import kafka.utils.{VerifiableProperties, CommandLineUtils, Logging}
+import kafka.server.{KafkaServer, KafkaServerStartable}
+import kafka.utils.{CommandLineUtils, Logging}
 import org.apache.kafka.common.utils.Utils
 
+import scala.collection.JavaConversions._
+
 object Kafka extends Logging {
 
   def getPropsFromArgs(args: Array[String]): Properties = {
@@ -55,9 +55,7 @@ object Kafka extends Logging {
   def main(args: Array[String]): Unit = {
     try {
       val serverProps = getPropsFromArgs(args)
-      val serverConfig = KafkaConfig.fromProps(serverProps)
-      KafkaMetricsReporter.startReporters(new VerifiableProperties(serverProps))
-      val kafkaServerStartable = new KafkaServerStartable(serverConfig)
+      val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)
 
       // attach shutdown handler to catch control-c
       Runtime.getRuntime().addShutdownHook(new Thread() {

Diff for: core/src/main/scala/kafka/server/KafkaServerStartable.scala

@@ -17,8 +17,17 @@
 
 package kafka.server
 
-import kafka.utils.Logging
+import java.util.Properties
 
+import kafka.metrics.KafkaMetricsReporter
+import kafka.utils.{VerifiableProperties, Logging}
+
+object KafkaServerStartable {
+  def fromProps(serverProps: Properties) = {
+    KafkaMetricsReporter.startReporters(new VerifiableProperties(serverProps))
+    new KafkaServerStartable(KafkaConfig.fromProps(serverProps))
+  }
+}
 
 class KafkaServerStartable(val serverConfig: KafkaConfig) extends Logging {
   private val server = new KafkaServer(serverConfig)