package com.liveramp.kafka_service.server;

import java.util.Properties;

import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import kafka.utils.Time;

public class KafkaBroker {

  private final Properties properties;
  private final Time time;

  KafkaBroker(Properties properties, Time time) {
    this.properties = properties;
    this.time = time;
  }

  public void start() {
    KafkaConfig config = new KafkaConfig(properties);
    KafkaServer server = new KafkaServer(config, time);
    server.startup();
  }

  public static void main(String[] args) {
    KafkaBroker broker = KafkaBrokerBuilder.create()
        .setZookeeperConnect("10.99.32.1:2181,10.99.32.14:2181,10.99.32.36:2181")
        .setPort(9092)
        .setDeleteTopicEnable(true)
        .setLogDirs("/tmp/kafka-logs")
        .build();
    broker.start();
  }
}

@@ -569,24 +569,30 @@
 args) %7B%0A    
+final 
 KafkaBroker 
@@ -830,24 +830,139 @@
 ();%0A
-    broker.start
+%0A    Thread thread = new Thread() %7B%0A      @Override%0A      public void run() %7B%0A        broker.start();%0A      %7D%0A    %7D;%0A    thread.run
 ();%0A

put broker in a thread