title: KafkaAvroDeserializer does not return SpecificRecord but returns GenericRecord
body: <p>My <code>KafkaProducer</code> is able to use <code>KafkaAvroSerializer</code> to serialize objects to my topic. However, <code>KafkaConsumer.poll()</code> returns deserialized <code>GenericRecord</code> instead of my serialized class.</p>
  
  <p>MyKafkaProducer</p>
  
  <pre><code> KafkaProducer&lt;CharSequence, MyBean&gt; producer;
      try (InputStream props = Resources.getResource("producer.props").openStream()) {
        Properties properties = new Properties();
        properties.load(props);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            io.confluent.kafka.serializers.KafkaAvroSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            io.confluent.kafka.serializers.KafkaAvroSerializer.class);
        properties.put("schema.registry.url", "http://localhost:8081");
  
        MyBean bean = new MyBean();
        producer = new KafkaProducer&lt;&gt;(properties);
        producer.send(new ProducerRecord&lt;&gt;(topic, bean.getId(), bean));
  </code></pre>
  
  <p>My KafkaConsumer</p>
  
  <pre><code> try (InputStream props = Resources.getResource("consumer.props").openStream()) {
        properties.load(props);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
        properties.put("schema.registry.url", "http://localhost:8081");
        consumer = new KafkaConsumer&lt;&gt;(properties);
      }
      consumer.subscribe(Arrays.asList(topic));
      try {
        while (true) {
          ConsumerRecords&lt;CharSequence, MyBean&gt; records = consumer.poll(100);
          if (records.isEmpty()) {
            continue;
          }
          for (ConsumerRecord&lt;CharSequence, MyBean&gt; record : records) {
            MyBean bean = record.value(); // &lt;-------- This is throwing a cast Exception because it cannot cast GenericRecord to MyBean
            System.out.println("consumer received: " + bean);
          }
        }
  </code></pre>
  
  <p><code>MyBean bean = record.value();</code> That line throws a cast Exception because it cannot cast GenericRecord to MyBean.</p>
  
  <p>I'm using <code>kafka-client-0.9.0.1</code>, <code>kafka-avro-serializer-3.0.0</code>.</p>