title: Spring Kafka with Confluent Kafka Avro Deserializer
body: <p>I am trying to use Spring Kafka with Confluent schema registry and Kafka Avro Deserializer. However there is no such thing in the Spring Kafka document.</p>
  
  <p>Following is what I do for my Spring Kafka consumer:</p>
  
  <pre><code>@Configuration
  @EnableKafka
  public class KafkaConfiguration {
  
    @Bean
    ConcurrentKafkaListenerContainerFactory&lt;String, Customer&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, Customer&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
      return factory;
    }
  
    @Bean
    public ConsumerFactory&lt;String, Customer&gt; consumerFactory() {
      return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
    }
  
    @Bean
    public Map&lt;String, Object&gt; consumerConfigs() {
      Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
      props.put("schema.registry.url", "http://127.0.0.1:8081");
      props.put("specific.avro.reader", "true");
  
  //    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // disable auto commit of offsets
      props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100"); // disable auto commit of offsets
      return props;
    }
  }
  </code></pre>
  
  <p>and </p>
  
  <pre><code>@Component
  @Slf4j
  public class KafkaConsumerService {
  
    private CustomerConverter customerConverter;
  
    public KafkaConsumerService(CustomerConverter customerConverter) {
      this.customerConverter = customerConverter;
    }
  
    @KafkaListener(id = "demo-consumer-1st-group", topics = "kafka-demo-avro")
    public void process(@Payload Customer customer,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List&lt;Integer&gt; partitions,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) List&lt;String&gt; topics,
                        @Header(KafkaHeaders.OFFSET) List&lt;Long&gt; offsets) {
      LOGGER.info("topic: {}, partition: {}, offset: {}", topics.get(0), partitions.get(0), offsets.get(0));
      CustomerModel customerModel = customerConverter.convertToModel(customer);
      LOGGER.info("customer: {}", customerModel);
    }
  }
  </code></pre>
  
  <p>My producer can produce Avro message successfully without any problem. But the error I got for consumer is:</p>
  
  <pre><code>Method [public void kafkademo.kafka.KafkaConsumerService.process(kafkademo.generated.avro.Customer)]
  Bean [kafkademo.kafka.KafkaConsumerService@238258a0]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [kafkademo.generated.avro.Customer] to [kafkademo.generated.avro.Customer] for GenericMessage [payload={"id": 7, "key": "key-789", "name": "kafka"}, headers={kafka_offset=6, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@7e280b0e, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=key-789, kafka_receivedPartitionId=0, kafka_receivedTopic=kafka-demo-avro, kafka_receivedTimestamp=1562593667254}], failedMessage=GenericMessage [payload={"id": 7, "key": "key-789", "name": "kafka"}, headers={kafka_offset=6, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@7e280b0e, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=key-789, kafka_receivedPartitionId=0, kafka_receivedTopic=kafka-demo-avro, kafka_receivedTimestamp=1562593667254}]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [kafkademo.generated.avro.Customer] to [kafkademo.generated.avro.Customer] for GenericMessage [payload={"id": 7, "key": "key-789", "name": "kafka"}, headers={kafka_offset=6, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@7e280b0e, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=key-789, kafka_receivedPartitionId=0, kafka_receivedTopic=kafka-demo-avro, kafka_receivedTimestamp=1562593667254}], failedMessage=GenericMessage [payload={"id": 7, "key": "key-789", "name": "kafka"}, headers={kafka_offset=6, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@7e280b0e, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=key-789, kafka_receivedPartitionId=0, kafka_receivedTopic=kafka-demo-avro, kafka_receivedTimestamp=1562593667254}]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1324) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1313) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1229) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1200) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1120) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:935) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:751) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:700) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      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.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [kafkademo.generated.avro.Customer] to [kafkademo.generated.avro.Customer] for GenericMessage [payload={"id": 7, "key": "key-789", "name": "kafka"}, headers={kafka_offset=6, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@7e280b0e, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=key-789, kafka_receivedPartitionId=0, kafka_receivedTopic=kafka-demo-avro, kafka_receivedTimestamp=1562593667254}], failedMessage=GenericMessage [payload={"id": 7, "key": "key-789", "name": "kafka"}, headers={kafka_offset=6, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@7e280b0e, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=key-789, kafka_receivedPartitionId=0, kafka_receivedTopic=kafka-demo-avro, kafka_receivedTimestamp=1562593667254}]
      at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:292) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:79) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:50) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1275) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1258) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1219) ~[spring-kafka-2.2.7.RELEASE.jar:2.2.7.RELEASE]
      ... 8 common frames omitted
  </code></pre>