Kafka high level API8 Consumer

import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit;   import org.apache.log4j.Logger; import com.jd.Seckill.UdpRecv.Utils.Config; import kafka.consumer.ConsumerConfig; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.consumer.Consumer;   public class KafkaConsume_HighAPI8 { private static Logger log = Logger.getLogger(KafkaConsume_HighAPI8.class); private final ConsumerConnector consumer; private final String topic; private ExecutorService executor;   private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) { Properties props = new Properties(); props.put("zookeeper.connect", a_zookeeper); props.put("group.id", a_groupId); props.put("zookeeper.session.timeout.ms", "400"); props.put("zookeeper.sync.time.ms", "200"); props.put("auto.commit.interval.ms", "1000");   return new ConsumerConfig(props); }   public KafkaConsume_HighAPI8(String a_zookeeper, String a_groupId, String a_topic) { consumer = Consumer.createJavaConsumerConnector(createConsumerConfig(a_zookeeper, a_groupId)); this.topic = a_topic; }   public void shutdown() { if (consumer != null) consumer.shutdown(); if (executor != null) executor.shutdown(); try { if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) { System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly"); } } catch (InterruptedException e) { System.out.println("Interrupted during shutdown, exiting uncleanly"); } }   public void run(int a_numThreads) { Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); topicCountMap.put(topic, new Integer(a_numThreads)); Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);   executor = Executors.newFixedThreadPool(a_numThreads);   int threadNumber = 0; for (final KafkaStream stream : streams) { executor.submit(new ConsumerTest(stream, threadNumber)); threadNumber++; } }   public void mainHandle() { String zooKeeper = Config.getM_strZkServers(); String groupId = "a"; String topic = Config.getM_strConsume8Topic(); int threads = 2;   KafkaConsume_HighAPI8 example = new KafkaConsume_HighAPI8(zooKeeper, groupId, topic); example.run(threads);   try { Thread.sleep(10000); } catch (InterruptedException e) { log.error(e); } example.shutdown(); } }