title: Apache Kafka Default Encoder Not Working
body: <p>I am using Kafka 0.8 beta, and I am just trying to mess around with sending different objects, serializing them using my own encoder, and sending them to an existing broker configuration. For now I am trying to get just DefaultEncoder working.</p>
  
  <p>I have the broker and everything setup and working for StringEncoder, but I am not able to get any other data type, including just pure byte[], to be sent and received by the broker. </p>
  
  <p>My code for the Producer is:</p>
  
  <pre><code>import kafka.javaapi.producer.Producer;
  import kafka.producer.KeyedMessage;
  import kafka.producer.ProducerConfig;
  
  import java.util.Date;
  import java.util.Properties;
  import java.util.Random;
  
  
  public class ProducerTest {
      public static void main(String[] args) {
          long events = 5;
          Random rnd = new Random();
          rnd.setSeed(new Date().getTime());
          Properties props = new Properties();
          props.setProperty("metadata.broker.list", "localhost:9093,localhost:9094");
          props.setProperty("serializer.class", "kafka.serializer.DefaultEncoder");
          props.setProperty("partitioner.class", "example.producer.SimplePartitioner");
          props.setProperty("request.required.acks", "1");
          props.setProperty("producer.type", "async");
          props.setProperty("batch.num.messages", "4");
  
          ProducerConfig config = new ProducerConfig(props);
          Producer&lt;byte[], byte[]&gt; producer = new Producer&lt;byte[], byte[]&gt;(config);
          for (long nEvents = 0; nEvents &lt; events; nEvents++) {
              byte[] a = "Hello".getBytes();
              byte[] b = "There".getBytes();
  
              KeyedMessage&lt;byte[], byte[]&gt; data = new KeyedMessage&lt;byte[], byte[]&gt;("page_visits", a, b);
              producer.send(data);
          }
          try {
              Thread.sleep(5000);
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
          producer.close();
      }
  }
  </code></pre>
  
  <p>I used the same SimplePartitioner as in the example given <a href="https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example" rel="nofollow">here</a>, and replacing all the byte arrays by Strings and changing the serializer to kafka.serializer.StringEncoder works perfectly. </p>
  
  <p>For reference, SimplePartitioner:</p>
  
  <pre><code>import kafka.producer.Partitioner;
  import kafka.utils.VerifiableProperties;
  
  public class SimplePartitioner implements Partitioner&lt;String&gt; {
      public SimplePartitioner (VerifiableProperties props) {
  
      }
  
      public int partition(String key, int a_numPartitions) {
          int partition = 0;
          int offset = key.lastIndexOf('.');
          if (offset &gt; 0) {
             partition = Integer.parseInt( key.substring(offset+1)) % a_numPartitions;
          }
         return partition;
    }
  
  }
  </code></pre>
  
  <p>What am I doing wrong?</p>