@@ -0,0 +1,2899 @@
+package com.test.groups;%0A %0Aimport kafka.consumer.ConsumerConfig;%0Aimport kafka.consumer.KafkaStream;%0Aimport kafka.javaapi.consumer.ConsumerConnector;%0A %0Aimport java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.Properties;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors;%0A %0Apublic class ConsumerGroupExample %7B%0A    private final ConsumerConnector consumer;%0A    private final String topic;%0A    private  ExecutorService executor;%0A %0A    public ConsumerGroupExample(String a_zookeeper, String a_groupId, String a_topic) %7B%0A        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(%0A                createConsumerConfig(a_zookeeper, a_groupId));%0A        this.topic = a_topic;%0A    %7D%0A %0A    public void shutdown() %7B%0A        if (consumer != null) consumer.shutdown();%0A        if (executor != null) executor.shutdown();%0A        try %7B%0A            if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) %7B%0A                System.out.println(%22Timed out waiting for consumer threads to shut down, exiting uncleanly%22);%0A            %7D%0A        %7D catch (InterruptedException e) %7B%0A            System.out.println(%22Interrupted during shutdown, exiting uncleanly%22);%0A        %7D%0A   %7D%0A %0A    public void run(int a_numThreads) %7B%0A        Map%3CString, Integer%3E topicCountMap = new HashMap%3CString, Integer%3E();%0A        topicCountMap.put(topic, new Integer(a_numThreads));%0A        Map%3CString, List%3CKafkaStream%3Cbyte%5B%5D, byte%5B%5D%3E%3E%3E consumerMap = consumer.createMessageStreams(topicCountMap);%0A        List%3CKafkaStream%3Cbyte%5B%5D, byte%5B%5D%3E%3E streams = consumerMap.get(topic);%0A %0A        // now launch all the threads%0A        //%0A        executor = Executors.newFixedThreadPool(a_numThreads);%0A %0A        // now create an object to consume the messages%0A        //%0A        int threadNumber = 0;%0A        for (final KafkaStream stream : streams) %7B%0A            executor.submit(new ConsumerTest(stream, threadNumber));%0A            threadNumber++;%0A        %7D%0A    %7D%0A %0A    private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) %7B%0A        Properties props = new Properties();%0A        props.put(%22zookeeper.connect%22, a_zookeeper);%0A        props.put(%22group.id%22, a_groupId);%0A        props.put(%22zookeeper.session.timeout.ms%22, %22400%22);%0A        props.put(%22zookeeper.sync.time.ms%22, %22200%22);%0A        props.put(%22auto.commit.interval.ms%22, %221000%22);%0A %0A        return new ConsumerConfig(props);%0A    %7D%0A %0A    public static void main(String%5B%5D args) %7B%0A        String zooKeeper = args%5B0%5D;%0A        String groupId = args%5B1%5D;%0A        String topic = args%5B2%5D;%0A        int threads = Integer.parseInt(args%5B3%5D);%0A %0A        ConsumerGroupExample example = new ConsumerGroupExample(zooKeeper, groupId, topic);%0A        example.run(threads);%0A %0A        try %7B%0A            Thread.sleep(10000);%0A        %7D catch (InterruptedException ie) %7B%0A %0A        %7D%0A        example.shutdown();%0A    %7D%0A%7D%0A

Create ConsumerGroupExample.java