package com.liveramp.kafka_service.broker;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Set;

import com.google.common.collect.Sets;
import kafka.admin.AdminUtils;
import kafka.admin.TopicCommand;
import kafka.api.TopicMetadata;
import kafka.common.TopicExistsException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.Iterator;

import com.liveramp.kafka_service.zookeeper.ZookeeperEnv;
import com.liveramp.kafka_service.zookeeper.ZookeeperClient;

public class TopicHelper {
  private static final Logger LOG = LoggerFactory.getLogger(TopicHelper.class);
  private static final int DEFAULT_PARTITIONS = 1;
  private static final int DEFAULT_REPLICATION_FACTOR = 1;

  private TopicHelper() {
    throw new AssertionError("Don't instantiate the class");
  }

  public static void createTopic(ZookeeperClient client, String topic) {
    createTopic(client, topic, DEFAULT_PARTITIONS, DEFAULT_REPLICATION_FACTOR);
  }

  public static void createTopic(ZookeeperClient client, String topic, int partitions, int replicationFactor) {
    String[] arguments = new String[]{
        "--create",
        "--topic", topic,
        "--partitions", String.valueOf(partitions),
        "--replication-factor", String.valueOf(replicationFactor)
    };
    TopicCommand.TopicCommandOptions options = new TopicCommand.TopicCommandOptions(arguments);
    try {
      TopicCommand.createTopic(client.get(), options);
    } catch (TopicExistsException e) {
      LOG.error("Topic {} already exists", topic);
    }
  }

  public static void deleteTopic(ZookeeperClient client, String topic) {
    try {
      AdminUtils.deleteTopic(client.get(), topic);
    } catch (ZkNodeExistsException e) {
      LOG.error("Topic {} has already been deleted", topic);
    }
  }

  public static TopicMetadata getDescription(ZookeeperClient client, String topic) {
    return AdminUtils.fetchTopicMetadataFromZk(topic, client.get());
  }

  public static Set<String> getAllTopics(ZookeeperClient client) {
    Set<String> topics = Sets.newHashSet();
    Iterator<String> iter = AdminUtils.fetchAllTopicConfigs(client.get()).keySet().iterator();
    while (iter.hasNext()) {
      topics.add(iter.next());
    }
    return topics;
  }

  public static void main(String[] args) throws IOException {
    ZookeeperClient client = new ZookeeperClient.Builder(ZookeeperEnv.getZKInstances()).build();
    BufferedReader br = new BufferedReader(new InputStreamReader(System.in));

    System.out.println("please enter the command:");
    String command = null;
    while ((command = br.readLine()) != null) {
      if (command.startsWith("create")) {
        String topic = command.split(" ")[1];
        createTopic(client, topic);
        System.out.println("created topic " + topic);
      } else if (command.equals("list")) {
        System.out.println(getAllTopics(client));
      } else if (command.startsWith("delete")) {
        String topic = command.split(" ")[1];
        deleteTopic(client, topic);
        System.out.println("deleted topic " + topic);
      } else if (command.startsWith("describe")) {
        String topic = command.split(" ")[1];
        System.out.println(getDescription(client, topic));
      } else if (command.equals("exit")) {
        break;
      } else {
        System.out.println("No such command: " + command);
      }
    }
    client.close();
  }

}

@@ -122,32 +122,61 @@
 utStreamReader;%0A
+import java.util.Properties;%0A
 import java.util
@@ -256,41 +256,8 @@
 ls;%0A
-import kafka.admin.TopicCommand;%0A
 impo
@@ -502,44 +502,42 @@
 amp.
-kafka_service.zookeeper.ZookeeperEnv
+java_support.logging.LoggingHelper
 ;%0Aim
@@ -594,16 +594,74 @@
 rClient;
+%0Aimport com.liveramp.kafka_service.zookeeper.ZookeeperEnv;
 %0A%0Apublic
@@ -872,16 +872,81 @@
 R = 1;%0A%0A
+  static %7B%0A    LoggingHelper.setLoggingProperties(%22topic%22);%0A  %7D%0A%0A
   privat
@@ -1307,100 +1307,64 @@
     
-S
 tr
-ing%5B%5D arguments = new String%5B%5D%7B%0A        %22--create%22,%0A        %22--topic%22, topic,%0A        %22--
+y %7B%0A      AdminUtils.createTopic(client.get(), topic, 
 part
@@ -1373,50 +1373,10 @@
 ions
-%22
 , 
-String.valueOf(partitions),%0A        %22--
 repl
@@ -1386,217 +1386,32 @@
 tion
--f
+F
 actor
-%22
 , 
-String.valueOf(replicationFactor)%0A    %7D;%0A    TopicCommand.TopicCommandOptions options = new TopicCommand.TopicCommandOptions(arguments);%0A    try %7B%0A      TopicCommand.createTopic(client.get(), options
+new Properties()
 );%0A 

use admin util for all topic operations