title: Apache Kafka Avro Deserialization: Unable to deserialize or decode Specific type message.
body: <p>I am trying to use Avro Serialize with Apache kafka for serialize/deserialize messages. I am create one producer, which is used to serialize specific type message and send it to the queue. When message is send successfully to the queue, our consumer pick the message and trying to process, but while trying we are facing an exception, for case bytes to specific object. The exception is as below: </p>
  
  <pre><code>[error] (run-main-0) java.lang.ClassCastException: org.apache.avro.generic.GenericData$Record cannot be cast to com.harmeetsingh13.java.avroserializer.Customer
  java.lang.ClassCastException: org.apache.avro.generic.GenericData$Record cannot be cast to com.harmeetsingh13.java.avroserializer.Customer
      at com.harmeetsingh13.java.consumers.avrodesrializer.AvroSpecificDeserializer.lambda$infiniteConsumer$0(AvroSpecificDeserializer.java:51)
      at java.lang.Iterable.forEach(Iterable.java:75)
      at com.harmeetsingh13.java.consumers.avrodesrializer.AvroSpecificDeserializer.infiniteConsumer(AvroSpecificDeserializer.java:46)
      at com.harmeetsingh13.java.consumers.avrodesrializer.AvroSpecificDeserializer.main(AvroSpecificDeserializer.java:63)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  </code></pre>
  
  <p>According to exception, we are using some inconenient way for read the data, below is our code: </p>
  
  <p>Kafka Producer Code: </p>
  
  <pre><code>static {
          kafkaProps.put("bootstrap.servers", "localhost:9092");
          kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
          kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
          kafkaProps.put("schema.registry.url", "http://localhost:8081");
          kafkaProducer = new KafkaProducer&lt;&gt;(kafkaProps);
      }
  
  
  public static void main(String[] args) throws InterruptedException, IOException {
          Customer customer1 = new Customer(1002, "Jimmy");
  
          Parser parser = new Parser();
          Schema schema = parser.parse(AvroSpecificProducer.class
                  .getClassLoader().getResourceAsStream("avro/customer.avsc"));
  
          SpecificDatumWriter&lt;Customer&gt; writer = new SpecificDatumWriter&lt;&gt;(schema);
          try(ByteArrayOutputStream os = new ByteArrayOutputStream()) {
              BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(os, null);
              writer.write(customer1, encoder);
              encoder.flush();
  
              byte[] avroBytes = os.toByteArray();
  
              ProducerRecord&lt;String, byte[]&gt; record1 = new ProducerRecord&lt;&gt;("CustomerSpecificCountry",
                      "Customer One 11 ", avroBytes
              );
  
              asyncSend(record1);
          }
  
          Thread.sleep(10000);
      }
  </code></pre>
  
  <p>Kafka Consumer Code: </p>
  
  <pre><code>static {
          kafkaProps.put("bootstrap.servers", "localhost:9092");
          kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
          kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
          kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountryGroup1");
          kafkaProps.put("schema.registry.url", "http://localhost:8081");
      }
  
      public static void infiniteConsumer() throws IOException {
          try(KafkaConsumer&lt;String, byte[]&gt; kafkaConsumer = new KafkaConsumer&lt;&gt;(kafkaProps)) {
              kafkaConsumer.subscribe(Arrays.asList("CustomerSpecificCountry"));
  
              while(true) {
                  ConsumerRecords&lt;String, byte[]&gt; records = kafkaConsumer.poll(100);
                  System.out.println("&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;" + records.count());
  
                  Schema.Parser parser = new Schema.Parser();
                  Schema schema = parser.parse(AvroSpecificDeserializer.class
                          .getClassLoader().getResourceAsStream("avro/customer.avsc"));
  
                  records.forEach(record -&gt; {
                      DatumReader&lt;Customer&gt; customerDatumReader = new SpecificDatumReader&lt;&gt;(schema);
                      BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(record.value(), null);
                      try {
                          System.out.println("&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;&gt;");
                          Customer customer = customerDatumReader.read(null, binaryDecoder);
                          System.out.println(customer);
                      } catch (IOException e) {
                          e.printStackTrace();
                      }
                  });
              }
  
          }
      }
  </code></pre>
  
  <p>Using consumer in console, we are successfully able to receive the message. So what is the way for decode message into our pojo files ?</p>