package nl.us2.cloudpelican.stormprocessor;

import storm.kafka.*;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.TimeZone;
import java.util.UUID;

/**
 * Created by robin on 07/06/15.
 */
public class Main {
    public static String ZOOKEEPER_NODES = "";
    public static String KAFKA_SPOUT = "kafka_spout";
    public static String MATCH_BOLT = "match_bolt";
    private static boolean isRunning = true;

    public static void main(String [] args) throws Exception
    {
        ArrayList<String> argList = new ArrayList<String>();
        for (String arg : args) {
            argList.add(arg);
        }

        // Config
        HashMap<String, String> settings = new HashMap<String, String>();
        for (String arg : argList) {
            String[] split = arg.split("=", 2);
            if (split.length == 2) {
                if (split[0].equals("-zookeeper")) {
                    Main.ZOOKEEPER_NODES = split[1];
                } else if (split[0].equals("-grep")) {
                    settings.put("match_regex", split[1]);
                } else if (split[0].equals("-topic")) {
                    settings.put("kafka_topic", split[1]);
                }
            }
        }
        System.out.println(settings);

        // Topology
        TopologyBuilder builder = new TopologyBuilder();
        int globalConcurrency = 1;

        // Time
        TimeZone.setDefault(TimeZone.getTimeZone("Etc/UTC"));

        // Read from kafka

        BrokerHosts hosts = new ZkHosts(ZOOKEEPER_NODES);
        SpoutConfig spoutConfig = new SpoutConfig(hosts, settings.get("kafka_topic"), "/" + settings.get("kafka_topic"), UUID.randomUUID().toString());
        spoutConfig.startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
        builder.setSpout(KAFKA_SPOUT, kafkaSpout, 3);

        // Match bolt
        builder.setBolt(MATCH_BOLT, new MatchBolt(settings.get("match_regex")), globalConcurrency * 4).localOrShuffleGrouping(KAFKA_SPOUT);

        // Debug on for testing
        Config conf = new Config();
        conf.setDebug(false);
        String topologyName = "cloudpelican_stormprocessor";
        if (argList.contains("-submit")) {
            conf.setNumWorkers(globalConcurrency);
            conf.setNumAckers(globalConcurrency);
            conf.setMaxSpoutPending(1000);
            System.out.println(topologyName);
            StormSubmitter.submitTopologyWithProgressBar(topologyName, conf, builder.createTopology());
        } else {
            //System.err.println("Disabled local mode");
            //System.exit(-1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(topologyName, conf, builder.createTopology());

            // Keep running until interrupt
            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run( ){
                    isRunning = false;
                }
            });
            while (isRunning) {
                Thread.sleep(100);
            }

            cluster.killTopology(topologyName);
            cluster.shutdown();
        }
    }
}

@@ -38,16 +38,73 @@
 essor;%0A%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A
 import s
@@ -117,16 +117,16 @@
 afka.*;%0A
-
 import b
@@ -717,16 +717,91 @@
  = true;
+%0A    private static final Logger LOG = LoggerFactory.getLogger(Main.class);
 %0A%0A    pu
@@ -1600,35 +1600,36 @@
     
-System.out.println(settings
+LOG.info(settings.toString()
 );%0A%0A
@@ -2839,54 +2839,8 @@
 0);%0A
-            System.out.println(topologyName);%0A
     
@@ -2960,96 +2960,8 @@
 e %7B%0A
-            //System.err.println(%22Disabled local mode%22);%0A            //System.exit(-1);%0A
     
@@ -3228,16 +3228,16 @@
 verride%0A
-
         
@@ -3264,16 +3264,63 @@
 run( )%7B%0A
+                    LOG.info(%22Shutting down%22);%0A
         

Use decent logging