package com.viadeo.axonframework.eventhandling.terminal.kafka;

import kafka.admin.TopicCommand;
import kafka.common.TopicExistsException;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.JavaConversions;

import java.util.List;

public class TopicStatement {

    private static final Logger LOGGER = LoggerFactory.getLogger(TopicStatement.class);

    private final String zkConnect;

    public TopicStatement(final String zkConnect) {
        this.zkConnect = zkConnect;
    }

    public void create(final String... topics) {
        final ZkClient zkClient = createZkClient();
        final List<String> existingTopics = getTopics(zkClient);

        try {
            for (final String topic:topics) {
                if ( ! (existingTopics.contains(topic) || exists(zkClient, topic)) ) {

                    String[] args = {
                            "--zookeeper", zkConnect,
                            "--partitions", "1",
                            "--replication-factor", "1",
                            "--create",
                            "--topic", topic
                    };

                    final TopicCommand.TopicCommandOptions opts = new TopicCommand.TopicCommandOptions(args);
                    opts.checkArgs();

                    try {
                        TopicCommand.createTopic(zkClient, opts);
                    } catch (TopicExistsException e) {
                        LOGGER.warn("unable to create topic, the topic '{}' is already defined", topic, e);
                    }
                }
            }
        } finally {
            zkClient.close();
        }
    }

    public void remove(final String... topics) {
        final ZkClient zkClient = createZkClient();

        try {
            for (String topic:topics) {
                if (exists(zkClient, topic)) {

                    String[] args = {
                            "--zookeeper", zkConnect,
                            "--delete",
                            "--topic", topic
                    };

                    TopicCommand.TopicCommandOptions opts = new TopicCommand.TopicCommandOptions(args);
                    opts.checkArgs();

                    try {
                        TopicCommand.alterTopic(zkClient, opts);
                    } catch (Throwable e) {
                        LOGGER.warn("unable to delete topic: '{}", topic, e);
                    }
                }
            }
        } finally {
            zkClient.close();
        }
    }

    public ZkClient createZkClient(){
        return new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer$.MODULE$);
    }

    public boolean exists(final ZkClient zkClient, final String topic) {
        return zkClient.exists(ZkUtils.getTopicPath(topic));
    }

    protected List<String> getTopics() {
        final ZkClient zkClient = createZkClient();
        try {
            return this.getTopics(zkClient);
        } finally {
            zkClient.close();
        }
    }

    protected List<String> getTopics(final ZkClient zkClient) {
        return JavaConversions.asList(ZkUtils.getAllTopics(zkClient));
    }

}

@@ -57,16 +57,55 @@
 kafka;%0A%0A
+import kafka.admin.DeleteTopicCommand;%0A
 import k
@@ -2098,48 +2098,8 @@
 ct,%0A
-                            %22--delete%22,%0A
     
@@ -2167,151 +2167,8 @@
 %7D;%0A%0A
-                    TopicCommand.TopicCommandOptions opts = new TopicCommand.TopicCommandOptions(args);%0A                    opts.checkArgs();%0A%0A
     
@@ -2205,32 +2205,38 @@
                 
+Delete
 TopicCommand.alt
@@ -2236,32 +2236,16 @@
 and.
-alterTopic(zkClient, opt
+main(arg
 s);%0A

Update the way to delete a topic on kafka