title: Simple Kafka Consumer Example not working
body: <p>I have a simple class to consume messages from a kafka server. The majority of codes are copied from the comments of org.apache.kafka.clients.consumer.KafkaConsumer.java.</p>
  
  <pre><code>public class Demo {
  
      public static void main(String[] args) {
          Properties props = new Properties();
          props.put("metadata.broker.list", "192.168.144.10:29092");
          props.put("group.id", "test");
          props.put("session.timeout.ms", "1000");
          props.put("enable.auto.commit", "true");
          props.put("auto.commit.interval.ms", "10000");
          KafkaConsumer&lt;byte[], byte[]&gt; consumer = new KafkaConsumer&lt;byte[], byte[]&gt;(props);
          consumer.subscribe("voltdbexportAUDIT", "voltdbexportTEST");
          boolean isRunning = true;
          while (isRunning) {
              Map&lt;String, ConsumerRecords&lt;byte[], byte[]&gt;&gt; records = consumer.poll(100);
              process(records);
          }
          consumer.close();
      }
  
      private static Map&lt;TopicPartition, Long&gt; process(Map&lt;String, ConsumerRecords&lt;byte[], byte[]&gt;&gt; records) {
          Map&lt;TopicPartition, Long&gt; processedOffsets = new HashMap&lt;&gt;();
          for (Map.Entry&lt;String, ConsumerRecords&lt;byte[], byte[]&gt;&gt; recordMetadata : records.entrySet()) {
              List&lt;ConsumerRecord&lt;byte[], byte[]&gt;&gt; recordsPerTopic = recordMetadata.getValue().records();
              for (int i = 0; i &lt; recordsPerTopic.size(); i++) {
                  ConsumerRecord&lt;byte[], byte[]&gt; record = recordsPerTopic.get(i);
                  // process record
                  try {
                      processedOffsets.put(record.topicAndPartition(), record.offset());
                  } catch (Exception e) {
                      e.printStackTrace();
                  }
              }
          }
          return processedOffsets;
      }
  }
  </code></pre>
  
  <p>I am using 'org.apache.kafka:kafka-clients:0.8.2.0'. it throws exception</p>
  
  <pre><code>Exception in thread "main" org.apache.kafka.common.config.ConfigException: Missing required configuration "key.deserializer" which has no default value.
      at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:124)
      at org.apache.kafka.common.config.AbstractConfig.&lt;init&gt;(AbstractConfig.java:48)
      at org.apache.kafka.clients.consumer.ConsumerConfig.&lt;init&gt;(ConsumerConfig.java:194)
      at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:430)
      at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:413)
      at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:400)
      at kafka.integration.Demo.main(Demo.java:26)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:497)
      at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)
  </code></pre>
  
  <p>How should I config the key.deserializer? </p>