title: when instantiate kafka producer with custom encoder, can't instantiate
body: <p>I am new to kafka. I want to use kafka producer/consumer to replace activeMQ(jms) in my spring project. What I need is a kafka producer publish my message object to a topic and a consumer subscribe it from the topic.</p>
  
  <p>first is my custom encoder,same thing for decoder(for my message class ConfigurationActionMsg):</p>
  
  <pre><code>@Component
  public class ConfigActionMessageEncoder implements Encoder&lt;ConfigurationActionMsg&gt; {
  
      public ConfigActionMessageEncoder() {
      /* This constructor must be present for successful compile. */
      }
      public ConfigActionMessageEncoder(VerifiableProperties verifiableProperties) {
      /* This constructor must be present for successful compile. */
      }
  
      @Override
      public byte[] toBytes(ConfigurationActionMsg actionMsg){
          return SerializationUtils.serialize(actionMsg);
      }}
  </code></pre>
  
  <p>Below is my configuration for processor and consumer</p>
  
  <pre><code>@Configuration
  @ComponentScan(basePackages = {"XXX"})
  public class KafkaConfig {
  @Bean
  public KafkaProducer&lt;String,ConfigurationActionMsg&gt; kafkaProducer(){
      Properties props = new Properties();
      props.put("zk.connect", "127.0.0.1:2181");
      props.put("bootstrap.servers", "localhost:9092");
      props.put("acks", "all");
      props.put("retries", 0);
      props.put("batch.size", 16384);
      props.put("linger.ms", 1);
      props.put("buffer.memory", 33554432);
      props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      props.put("value.serializer", "com.atlas.configengine2.XXX.ConfigActionMessageEncoder");
  
      return new KafkaProducer&lt;&gt;(props);
  }
  
  @Bean
  public KafkaConsumer&lt;String, ConfigurationActionMsg&gt; kafkaConsumer(){
  
      Properties props = new Properties();
      props.put("zk.connect", "127.0.0.1:2181");
      props.put("bootstrap.servers", "localhost:9092");
      //We should only have one process running for consumer
      props.put("group.id", "resolverActionTrigger");
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      props.put("value.deserializer", "com.atlas.configengine2.XXX.ConfigActionMessageDecoder");
      KafkaConsumer&lt;String, ConfigurationActionMsg&gt; consumer = new KafkaConsumer&lt;&gt;(props);
      consumer.subscribe(Arrays.asList("configAction"));
      return consumer;
  }}
  </code></pre>
  
  <p>I am not sure if this is the proper way to instantiate a producer/consumer. But this way is not working. Because my kafkaProducer couldn't be instantiated. </p>
  
  <p>some debug info:</p>
  
  <blockquote>
    <p>Caused by: org.apache.kafka.common.KafkaException:
    com.atlas.configengine2.jms.ConfigActionMessageEncoder is not an
    instance of org.apache.kafka.common.serialization.Serializer</p>
  </blockquote>
  
  <p>But I am not sure whether that is the only problem? And how to write the custom encoder then?</p>