title: How to write a KafkaAvro Serde for GenericData.Record
body: <p>I'm using Kafka 0.10.2 and Avro for the serialization of my messages, both for the key and for the value data.
  Now I would like to use Kafka Streams but I'm stuck trying to write the <code>Serde</code> class for the <code>GenericData.Record</code> class.</p>
  
  <pre><code>import org.apache.avro.generic.GenericData.Record;
  import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
  import io.confluent.kafka.serializers.KafkaAvroDeserializer;
  import io.confluent.kafka.serializers.KafkaAvroSerializer;
  [...]
  
  public final class KafkaAvroSerde implements Serde&lt;Record&gt; {
  
      private final Serde&lt;Record&gt; inner;
  
      public KafkaAvroSerde() {
          // Here I get the error
          inner = Serdes.serdeFrom(new KafkaAvroSerializer(), new KafkaAvroDeserializer());
      }
  
      public KafkaAvroSerde(SchemaRegistryClient client) {
          this(client, Collections.emptyMap());
      }
  
      public KafkaAvroSerde(SchemaRegistryClient client, Map&lt;String, ?&gt; props) {
          // Here I get the error
          inner = Serdes.serdeFrom(new KafkaAvroSerializer(client, props), new KafkaAvroDeserializer(client, props));
      }
  
      @Override
      public Serializer&lt;Record&gt; serializer() {
          return inner.serializer();
      }
  
      @Override
      public Deserializer&lt;Record&gt; deserializer() {
          return inner.deserializer();
      }
  
      @Override
      public void configure(Map&lt;String, ?&gt; configs, boolean isKey) {
          inner.serializer().configure(configs, isKey);
          inner.deserializer().configure(configs, isKey);
      }
  
      @Override
      public void close() {
          inner.serializer().close();
          inner.deserializer().close();
      }
  
  }
  </code></pre>
  
  <p>This is the error I'm getting at the commented lines</p>
  
  <pre><code>Type mismatch: cannot convert from Serde&lt;Object&gt; to Serde&lt;GenericData.Record&gt;
  </code></pre>
  
  <p>I need to define the Serde class for the <code>GenericData.Record</code> (and not for a specific POJO of mine) because I can have different object structures on the same channel, so the deserializer should return me the <code>GenericData</code> (and I will populate the right POJOs after this step).</p>
  
  <p>How would you make things done? 
  Thank you</p>