title: Generic type handling in Avro encoder/decoder
body: <p>I am using <code>spring-intergration-kafka</code> to send and receive messages from kafka. The message object is based on a generic type. Base class looks like this</p>
  
  <pre><code>public abstract class AbstractMessage&lt;T&gt; implements Serializable {
   ...
  }
  </code></pre>
  
  <p>I would like to be able to send and receive different implementations of this class. I am using the avro reflection based encoders and decoders</p>
  
  <pre><code>&lt;bean id="kafkaMessageEncoder"
      class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder"&gt;
      &lt;constructor-arg value="com.....model.AbstractMessage" /&gt;
  &lt;/bean&gt;
  
  &lt;bean id="kafkaMessageDecoder"
      class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder"&gt;
      &lt;constructor-arg value="com.....model.AbstractMessage" /&gt;
  &lt;/bean&gt;
  </code></pre>
  
  <p>This fails with the error</p>
  
  <blockquote>
    <p>AvroTypeException: Unknown type: T</p>
  </blockquote>
  
  <p>This makes sense because Avro cant figure out the generic type specified in the AbstractMessage class but then I decided to use custom encoder and decoder. </p>
  
  <pre><code>public class MessageEncoder implements Encoder&lt;Object&gt; {
  
  @Override
  public byte[] toBytes(Object object) {
      try {
          ByteArrayOutputStream baos = new ByteArrayOutputStream();
          ObjectOutputStream oos = new ObjectOutputStream(baos);
          oos.writeObject(object);
          oos.flush();
          oos.close();
          return baos.toByteArray();
      } catch (Exception e) {
          e.printStackTrace();
      }
      return "".getBytes();
  }
  
  }
  
  public class MessageDecoder implements Decoder&lt;Object&gt; {
  
  @Override
  public Object fromBytes(byte[] bs) {
      try {
          ObjectInputStream bais = new ObjectInputStream(new ByteArrayInputStream(bs));
          AbstractMessage&lt;?&gt; message = (AbstractMessage&lt;?&gt;) bais.readObject();
          return message;
      } catch (Exception e) {
          e.printStackTrace();
      }
      return null;
  }
  }
  </code></pre>
  
  <p>And this works fine.</p>
  
  <p>I guess my question is that I am not really doing anything special here. Why cant avro do the same thing and is there a different avro d/encoder that can serialize/deserialize my generic type objects.</p>
  
  <p>Thanks</p>