title: Avro class cannot be cast to class org.springframework.messaging.Message
body: <p>I am consuming Avro data coming from Debezium</p>
  <p>I made the kafka consumer as follows:</p>
  <ol>
  <li><p>The Java POJO</p>
  <pre><code>import lombok.Data;
  
  @Data
  public class Shop {
  
   Long shopId;
   Double latitude, longitude;
   String name;
   String phoneNumber;
   String placeId;
   double rating;
   String website;
   int addressId;
   String closingHours;
   String email;
   int maxAttendance;
   String opening_hours;
   String businessHours;
   String closeDay;
   String description;
   boolean open;
   String setWeekendBusinessHours;
   Long userShopId;
  }
  </code></pre>
  </li>
  <li><p>Avro Message Format</p>
  </li>
  </ol>
  <pre><code>     {
           &quot;type&quot;: &quot;record&quot;,
           &quot;name&quot;: &quot;ShopMessage&quot;,
           &quot;namespace&quot;: &quot;com.example.kafka.avro&quot;,
           &quot;fields&quot;: [
           {
            &quot;name&quot;: &quot;SHOP_ID&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;long&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;LATITUDE&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;double&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;LONGITUDE&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;double&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;NAME&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;PHONENUMBER&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;PLACEID&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;RATING&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;double&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;WEBSITE&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;ADDRESSID&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;int&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;CLOSINGHOUR&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;EMAIL&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;MAXATTENDANCE&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;int&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;OPENINGHOURS&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;BUSINESSHOURS&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;CLOSEDAY&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;DESCRIPTION&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;ISOPEN&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;boolean&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;WEEKENDBUSINESSHOURS&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;string&quot;
            ],
            &quot;default&quot;: null
           },
           {
            &quot;name&quot;: &quot;USERSHOPID&quot;,
            &quot;type&quot;: [
              &quot;null&quot;,
              &quot;long&quot;
            ],
            &quot;default&quot;: null
           }
          ]
         }
  </code></pre>
  <ol start="3">
  <li>ShopConsumer</li>
  </ol>
  <pre><code>@Component
  public class ShopConsumer {
  
      private final ShopMapper shopMapper;
      private final Logger log = LogManager.getLogger(ShopConsumer.class);
  
      public ShopConsumer(ShopMapper shopMapper) {
          this.shopMapper = shopMapper;
      }
  
      @KafkaListener(
              groupId = &quot;${spring.kafka.consumer.group-id}&quot;,
              topics = &quot;${spring.kafka.consumer.topic}&quot;
      )
      public void listen(List&lt;Message&lt;ShopMessage&gt;&gt; messages, Acknowledgment ack){
          log.info(&quot;Received batch of messages with size: {}&quot;, messages.size());
          List&lt;Shop&gt; shops = messages.stream()
                  .peek(this::logMessageReceived)
                  .map(message -&gt; shopMapper.toChange(message.getPayload()))
                  .collect(Collectors.toList());
  
          //do remove redis cache
          ack.acknowledge();
      }
  
      private void logMessageReceived(Message&lt;ShopMessage&gt; message) {
          log.info(&quot;Received shopId {} with a name of '{} and place id {}', partition={}, offset={}&quot;,
                  message.getPayload().getSHOPID(),
                  message.getPayload().getNAME(),
                  message.getPayload().getPLACEID(),
                  message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID),
                  message.getHeaders().get(KafkaHeaders.OFFSET));
      }
  
  </code></pre>
  <ol start="4">
  <li>Consumer Config - ShopConsumerConfig.java</li>
  </ol>
  <pre><code>@EnableKafka
  @Configuration
  public class ShopsConsumerConfig {
  
      private final KafkaProperties kafkaProperties;
  
      public ShopsConsumerConfig(KafkaProperties kafkaProperties) {
          this.kafkaProperties = kafkaProperties;
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, ShopMessage&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, ShopMessage&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          factory.setBatchListener(false);
          factory.setConcurrency(kafkaProperties.getListener().getConcurrency());
          factory.getContainerProperties().setAckMode(kafkaProperties.getListener().getAckMode());
          return factory;
      }
  
      @Bean
      public ConsumerFactory&lt;String, ShopMessage&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
      }
  
      @Bean
      public Map&lt;String, Object&gt; consumerConfigs() {
          Map&lt;String, Object&gt; props = kafkaProperties.buildConsumerProperties();
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SpecificAvroWithSchemaDeserializer.class);
  
          props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, kafkaProperties.getProperties().get(&quot;schema-registry-url&quot;));
          props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
          props.put(SpecificAvroWithSchemaDeserializer.AVRO_VALUE_RECORD_TYPE, ShopMessage.class);
          return props;
      }
  
  }
  
  </code></pre>
  <ol start="5">
  <li>Schema Deserializer</li>
  </ol>
  <pre><code>public class SpecificAvroWithSchemaDeserializer extends AbstractKafkaAvroDeserializer implements Deserializer&lt;Object&gt; {
  
      public static final String AVRO_KEY_RECORD_TYPE = &quot;avro.key.record.type&quot;;
      public static final String AVRO_VALUE_RECORD_TYPE = &quot;avro.value.record.type&quot;;
  
      private Schema readerSchema;
  
      public SpecificAvroWithSchemaDeserializer() { }
  
      @Override
      public void configure(Map&lt;String, ?&gt; configs, boolean isKey) {
          this.configure(new KafkaAvroDeserializerConfig(configs));
          readerSchema = getSchema(getRecordClass(configs, isKey));
      }
  
      private Class&lt;?&gt; getRecordClass(Map&lt;String, ?&gt; configs, boolean isKey) {
          String configsKey = isKey ? AVRO_KEY_RECORD_TYPE : AVRO_VALUE_RECORD_TYPE;
          Object configsValue = configs.get(configsKey);
  
          if (configsValue instanceof Class) {
              return (Class&lt;?&gt;) configsValue;
          } else if (configsValue instanceof String) {
              String recordClassName = (String) configsValue;
              try {
                  return Class.forName(recordClassName);
              } catch (ClassNotFoundException e) {
                  throw new IllegalArgumentException(String.format(&quot;Unable to find the class '%s'&quot;, recordClassName));
              }
          } else {
              throw new IllegalArgumentException(
                      String.format(&quot;A class or a string must be informed into ConsumerConfig properties: '%s' and/or '%s'&quot;,
                              AVRO_KEY_RECORD_TYPE, AVRO_VALUE_RECORD_TYPE));
          }
      }
  
      private Schema getSchema(Class&lt;?&gt; targetType) {
          try {
              Field field = targetType.getDeclaredField(&quot;SCHEMA$&quot;);
              return (Schema) field.get(null);
          } catch (NoSuchFieldException | IllegalAccessException e) {
              throw new IllegalArgumentException(
                      String.format(&quot;Unable to get Avro Schema from the class '%s'&quot;, targetType.getName()), e);
          }
      }
  
      @Override
      public Object deserialize(String topic, byte[] bytes) {
          return super.deserialize(bytes, readerSchema);
      }
  
      @Override
      public void close() {
      }
  
  
  }
  </code></pre>
  <ol start="6">
  <li>Mapper Class</li>
  </ol>
  <pre><code>@Mapper(componentModel = &quot;spring&quot;)
  public interface ShopMapper {
  
      default Shop toChange(ShopMessage shopMessage){
          if(shopMessage == null){
              return null;
          }
          Shop shop = new Shop();
          shop.setDescription(shopMessage.getDESCRIPTION().toString());
          shop.setMaxAttendance(shopMessage.getMAXATTENDANCE());
          shop.setSetWeekendBusinessHours(shopMessage.getWEEKENDBUSINESSHOURS().toString());
          shop.setOpen(shopMessage.getISOPEN());
          shop.setWebsite(shopMessage.getWEBSITE().toString());
          shop.setRating(shopMessage.getRATING());
          shop.setLatitude(shopMessage.getLATITUDE());
          shop.setLongitude(shopMessage.getLONGITUDE());
          shop.setCloseDay(shopMessage.getCLOSEDAY().toString());
          shop.setBusinessHours(shopMessage.getBUSINESSHOURS().toString());
          shop.setPhoneNumber(shopMessage.getPHONENUMBER().toString());
          shop.setEmail(shopMessage.getEMAIL().toString());
          shop.setPlaceId(shopMessage.getPLACEID().toString());
  
          return shop;
      }
  
  }
  </code></pre>
  <p>Configuration is present on the <code>application.properties</code> file but during message consumption, Spring throws me an error of</p>
  <pre><code>Caused by: java.lang.ClassCastException: class com.example.kafka.avro.ShopMessage cannot be cast to class org.springframework.messaging.Message (com.example.kafka.avro.ShopMessage and org.springframework.messaging.Message are in unnamed module of loader 'app')
  </code></pre>
  <p>Could someone give me a correct direction to fix this issue, please? Looks like casting from POJO from Avro is having the issue but I am not able to find the root.</p>
  <p>Thanks in advance.</p>
  <hr />
  <p><strong>Update</strong></p>
  <p>After few attempt, it looks that the issue on the above error is due to casting from a single message to list of messages. I changed the listener function as below.</p>
  <pre><code>public void listen(ConsumerRecord&lt;Integer ,?&gt; messages, Acknowledgment ack){
          //log.info(&quot;Received batch of messages with size: {}&quot;, messages.size());
          log.info(messages.key());
          log.info(messages.value());
    
          ack.acknowledge();
      }
  </code></pre>
  <p>and getting a value from Kafka topic.</p>
  <pre><code>{&quot;before&quot;: {&quot;id&quot;: 6, &quot;latitude&quot;: 2.921318, &quot;longitude&quot;: 101.655938, &quot;name&quot;: &quot;XYZ&quot;, &quot;phone_number&quot;: &quot;+12345678&quot;, &quot;place_id&quot;: &quot;P007&quot;, &quot;rating&quot;: 5.0, &quot;type&quot;: &quot;Food&quot;, &quot;website&quot;: &quot;https://xyz.me&quot;, &quot;address_id&quot;: 5, &quot;closing_hours&quot;: null, &quot;email&quot;: &quot;info@xyz.me&quot;, &quot;max_attendance&quot;: 11, &quot;opening_hours&quot;: null, &quot;business_hours&quot;: &quot;09-18&quot;, &quot;close_day&quot;: &quot;Saturday&quot;, &quot;description&quot;: &quot;Some Dummy&quot;, &quot;is_open&quot;: true, &quot;weekend_business_hours&quot;: &quot;08-12&quot;, &quot;user_shop_id&quot;: 0}, &quot;after&quot;: {&quot;id&quot;: 6, &quot;latitude&quot;: 2.921318, &quot;longitude&quot;: 101.655938, &quot;name&quot;: &quot;XYZ - edited&quot;, &quot;phone_number&quot;: &quot;+12345678&quot;, &quot;place_id&quot;: &quot;P007&quot;, &quot;rating&quot;: 5.0, &quot;type&quot;: &quot;Food 2&quot;, &quot;website&quot;: &quot;https://xyz.me&quot;, &quot;address_id&quot;: 5, &quot;closing_hours&quot;: null, &quot;email&quot;: &quot;info@xyz.me&quot;, &quot;max_attendance&quot;: 11, &quot;opening_hours&quot;: null, &quot;business_hours&quot;: &quot;09-18&quot;, &quot;close_day&quot;: &quot;Saturday&quot;, &quot;description&quot;: &quot;Some dummy&quot;, &quot;is_open&quot;: true, &quot;weekend_business_hours&quot;: &quot;08-12&quot;, &quot;user_shop_id&quot;: 0}, &quot;source&quot;: {&quot;version&quot;: &quot;1.6.0.Final&quot;, &quot;connector&quot;: &quot;mysql&quot;, &quot;name&quot;: &quot;bookingdev_sqip_local&quot;, &quot;ts_ms&quot;: 1629267837000, &quot;snapshot&quot;: &quot;false&quot;, &quot;db&quot;: &quot;booking&quot;, &quot;sequence&quot;: null, &quot;table&quot;: &quot;shop&quot;, &quot;server_id&quot;: 1, &quot;gtid&quot;: null, &quot;file&quot;: &quot;mysql-bin.000044&quot;, &quot;pos&quot;: 26432, &quot;row&quot;: 0, &quot;thread&quot;: null, &quot;query&quot;: null}, &quot;op&quot;: &quot;u&quot;, &quot;ts_ms&quot;: 1629267836453, &quot;transaction&quot;: null}
  </code></pre>
  <p>apart of that, I also removed the custom deserializer and custom POJO as schema is already installed on the schema registry.</p>
  <p>Now the question remains, how do I obtain debezium's schema generated from schema-registry and convert the message to the correct Java POJO to be executed further?</p>
  <hr />
  <p><strong>Update 19.08.2021</strong></p>
  <p>After discussion with @OneCricketeer, I made adjustment on the logic for Consumer as below</p>
  <pre><code> public void listen(ConsumerRecord&lt;Integer, GenericRecord&gt; messages, Acknowledgment ack) throws JsonProcessingException {
         
          log.info(messages.key());
          log.info(messages.value());
  
        
  
          Shop shop = new ObjectMapper().readValue(messages.value().get(&quot;after&quot;).toString(), Shop.class);
          log.info(&quot;NEW VALUE #####-&gt; &quot; + shop.getName());
          
          //other logic here.
  
          ack.acknowledge();
      }
  </code></pre>
  <p>But I got another error:</p>
  <pre><code>java.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer
      at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:194) ~[spring-kafka-2.7.0.jar:2.7.0]
      at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:112) ~[spring-kafka-2.7.0.jar:2.7.0]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1598) ~[spring-kafka-2.7.0.jar:2.7.0]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1210) ~[spring-kafka-2.7.0.jar:2.7.0]
      at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
      at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
      at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
  Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition bookingdev_sqip_local.booking.shop-0 at offset 0. If needed, please seek past the record to continue consumption.
  Caused by: org.apache.kafka.common.errors.SerializationException: Could not find class bookingdev_sqip_local.booking.shop.Key specified in writer's schema whilst finding reader's schema for a SpecificRecord.
  </code></pre>
  <p>Checked the Schema-Registry Debezium created two subjects - one for key and one for value.</p>
  <pre><code>[&quot;bookingdev_sqip_local.booking.shop-value&quot;,&quot;bookingdev_sqip_local.booking.shop-key&quot;]
  </code></pre>
  <p>Looks like the error due to unable to map the schema for the key.</p>