title: Kafka Consumer in Java
body: <p>So I am learning Kafka currently and have attempted to duplicate the examples provided from Apache <a href="https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example" rel="nofollow">here</a>. This is example code for the consumer and I have written it in java just as shown. When I attempt to execute the file however I run into some issues. I am able to get the file to compile but it will not run properly.</p>
  
  <p>I am executing the program with the following line without the quotations, "java TestConsumer localhost:2181 group1 test 4" This passes the 4 arguments necessary in the example code. I am provided with the following error though when I execute this command.</p>
  
  <pre><code>Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/log4j/Category
      at kafka.utils.VerifiableProperties.&lt;init&gt;(Unknown Source)
      at kafka.consumer.ConsumerConfig.&lt;init&gt;(Unknown Source)
      at TestConsumer.ConsumerProps(TestConsumer.java:69)
      at TestConsumer.&lt;init&gt;(TestConsumer.java:31)
      at TestConsumer.main(TestConsumer.java:97)
      Caused by: java.lang.ClassNotFoundException: org.apache.log4j.Category
      at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
      ... 5 more
  </code></pre>
  
  <p>I have tried going in an manually replacing the arguments with the necessary values and attempting to execute that way but I am given a different issue. Below is the error message along with the code I'm using just in case I screwed something up from the example provided. If anyone can help me out I would be incredibly appreciative since I am attempting to write my own consumer to test with parsing given information, etc. Thanks</p>
  
  <pre><code>log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
  log4j:WARN Please initialize the log4j system properly.
  Exception in thread "main" java.lang.NoClassDefFoundError: org/I0Itec/zkclient/IZkStateListener
  at kafka.javaapi.consumer.ZookeeperConsumerConnector.&lt;init&gt;(Unknown Source)
  at kafka.javaapi.consumer.ZookeeperConsumerConnector.&lt;init&gt;(Unknown Source)
  at kafka.consumer.Consumer$.createJavaConsumerConnector(Unknown Source)
  at kafka.consumer.Consumer.createJavaConsumerConnector(Unknown Source)
  at TestConsumer.&lt;init&gt;(TestConsumer.java:31)
  at TestConsumer.main(TestConsumer.java:97)
  Caused by: java.lang.ClassNotFoundException: org.I0Itec.zkclient.IZkStateListener
  at java.net.URLClassLoader.findClass(Unknown Source)
  at java.lang.ClassLoader.loadClass(Unknown Source)
  at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
  at java.lang.ClassLoader.loadClass(Unknown Source)
  ... 6 more
  </code></pre>
  
  <p><div class="snippet" data-lang="js" data-hide="false">
  <div class="snippet-code">
  <pre class="snippet-code-js lang-js prettyprint-override"><code>/*
   * Test Consumer to gather input from
   * a Producer. Attempt to perform functions
   * from the produced data
  */
  
  
  // Kafka API
  import kafka.consumer.ConsumerConfig;
  import kafka.consumer.KafkaStream;
  import kafka.javaapi.consumer.ConsumerConnector;
  
  import java.util.Map;
  import java.util.HashMap;
  import java.util.Properties;
  import java.util.List;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.TimeUnit;
  
  
  public class TestConsumer{
   
   private final ConsumerConnector consumer;
   private final String topic;
   private ExecutorService executor;
   
   
   // CONSTRUCTOR
   public TestConsumer(String zookeeper, String groupid, String aTopic){
    consumer = kafka.consumer.Consumer.createJavaConsumerConnector(ConsumerProps(zookeeper, groupid));
    this.topic = aTopic;
   }
   // END CONSTRUCTOR
   
   
   // RUN FUNCTION
   public void run(int threads){
    Map&lt;String, Integer&gt; topicMap = new HashMap&lt;String, Integer&gt;();
    topicMap.put(topic, new Integer(threads));
    Map&lt;String, List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt;&gt; consumerMap = consumer.createMessageStreams(topicMap);
    List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt; streams = consumerMap.get(topic);
    
    executor = Executors.newFixedThreadPool(threads); // process threads
    
    int numThread = 0; // thread counter for consumption
    
    // consumer all messages
    for(final KafkaStream stream : streams){
     executor.submit(new TestConsumerRun(stream, numThread));
     numThread ++;
    }
   }
   // END RUN FUNCTION
   
   
   // CREATE PROPERTIES FUNCTION
   private static ConsumerConfig ConsumerProps(String zookeeper, String groupid){
    
    Properties properties = new Properties(); // config properties file
    
    properties.put("zookeeper.connect", zookeeper);
    properties.put("group.id", groupid);
    properties.put("zookeeper.session.timeout.ms", "400");
    properties.put("zookeeper.sync.time.ms", "200");
    properties.put("auto.commit.interval.ms", "1000");
    properties.put("auto.offset.reset", "smallest");
    
    return new ConsumerConfig(properties);
   }
   // END CREATE PROPERTIES FUNCTION
   
   
   // SHUTDOWN FUNCTION
   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");
    }
   }
   // END SHUTDOWN FUNCTION
   
   
   // MAIN FUNCTION
   public static void main(String[] args){
    String zookeeper = args[0];
    String groupid = args[1];
    String topic = args[2];
    int threads = Integer.parseInt(args[3]);
    
    TestConsumer test = new TestConsumer(zookeeper, groupid, topic); // send information to constructor
    test.run(threads); // pass threads for iteration
    
    try{
     Thread.sleep(10000);
    } catch (InterruptedException ie){
    }
    
    test.shutdown(); // close program
   } 
   // END MAIN FUNCTION
   
  }</code></pre>
  </div>
  </div>
  </p>
  
  <p><div class="snippet" data-lang="js" data-hide="false">
  <div class="snippet-code">
  <pre class="snippet-code-js lang-js prettyprint-override"><code>/*
   * Test Consumer to gather input from
   * a Producer. Attempt to perform functions
   * from the produced data
  */
  
  // Kafka API
  import kafka.consumer.ConsumerIterator;
  import kafka.consumer.KafkaStream;
  
  
  public class TestConsumerRun implements Runnable{
   
   private KafkaStream aStream;
   private int aThread;
   
   
   // CONSTRUCTOR
   public TestConsumerRun(KafkaStream stream, int thread){
    aStream = stream; // set stream from main read
    aThread = thread; // set thread from main read
   }
   // END CONSTRUCTOR
   
   
   // RUN FUNCTION
   public void run(){
    
    ConsumerIterator&lt;byte[], byte[]&gt; iterator = aStream.iterator();  // used to check throughout the list continiously
    
    while(iterator.hasNext())
     System.out.println("Thread " + aThread + ": " + new String(iterator.next().message()));
    System.out.println("Shutting down Thread: " + aThread);
    
   }
   // END RUN FUNCTION
  }</code></pre>
  </div>
  </div>
  </p>