title: Serde class for AVRO primitive type
body: <p>I'm writing a Kafka stream app in Java that takes input topics created by a connector that uses the schema registry and avro for both the key and value converter. The connector produces the following schemas:</p>
  
  <pre><code>key-schema: "int"
  value-schema:{
  "type": "record",
  "name": "User",
  "fields": [
      {"name": "firstname", "type": "string"},
      {"name": "lastname",  "type": "string"}
  ]}
  </code></pre>
  
  <p>Actually, there are several topics, the key-schema is always "int" and the value-schema is always a record of some kind (User, Product, etc). My code contains the following definitions </p>
  
  <pre><code>Map&lt;String, String&gt; serdeConfig = Collections.singletonMap("schema.registry.url", schemaRegistryUrl);
  
  Serde&lt;User&gt; userSerde = new SpecificAvroSerde&lt;&gt;();
  userSerde.configure(serdeConfig, false);
  </code></pre>
  
  <p>At first I tried consuming the topic with something like 
  <code>Consumed.with(Serdes.Integer(), userSerde);</code> but that did not work because Serdes.Integer() expects integers to be encoded using 4 bytes but avro uses a variable length encoding. Using <code>Consumed.with(Serdes.Bytes(), userSerde);</code> worked but I really wanted int and not bytes so I changed my code to this</p>
  
  <pre><code>KafkaAvroDeserializer keyDeserializer = new KafkaAvroDeserializer()
  KafkaAvroSerializer keySerializer = new KafkaAvroSerializer();
  keyDeserializer.configure(serdeConfig, true); 
  keySerializer.configure(serdeConfig, true);
  Serde&lt;Integer&gt; keySerde = (Serde&lt;Integer&gt;)(Serde)Serdes.serdeFrom(keySerializer, keyDeserializer);
  </code></pre>
  
  <p>This made the compiler produce a warning (it doesn't like the <code>(Serde&lt;Integer&gt;)(Serde)</code> casting) but it allows me to use </p>
  
  <p><code>Consumed.with(keySerde, userSerde);</code> and get an integer as the key. This works just fine and my app is behaving as expected (great!!!). But now I want to define default serde for the key/value and I cannot get it to work.</p>
  
  <p>Setting the default value serde is simple:</p>
  
  <pre><code>streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
  </code></pre>
  
  <p>However I cannot figure out how to define the default key serde.</p>
  
  <p>I tried</p>
  
  <ol>
  <li><code>streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, keySerde.getClass().getName());</code>
  Produces runtime error: Could not find a public no-argument constructor for org.apache.kafka.common.serialization.Serdes$WrapperSerde</li>
  <li><code>streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);</code>
  Produces runtime error: java.lang.Integer cannot be cast to org.apache.avro.specific.SpecificRecord</li>
  </ol>
  
  <p>What am I missing?
  Thanks.</p>