title: Spring Boot Kafka - consumer custom JsonDeserializer with type info in yaml/properties, without @Bean
body: <p>Is it possible to pass type info to the deserializer of Kafka consumer without defining a <code>@Configuration</code> class with <code>@Bean</code> of <code>consumerFactory()</code> and <code>containerFactory</code>?</p>
  
  <p>In the <code>@Bean</code> method I have to put all configs I already have in yml file again into a map and pass it to the constructor of factories, but I think it is an overhead. I want to find a way to keep all configs in yaml/properties. (And I think it is cleaner to put configs into yaml, instead of in code)</p>
  
  <p>I want to specify type for each consumer/each methods annotated with <code>@KafkaListener</code> because I will receive distinct JSONs, mapped to its corresponding DTOs, so a universal <code>spring.json.value.default.type</code> is not applicable here.(And I think it is ugly)</p>
  
  <p>I am testing with my code(Spring Kafka Test with an <code>EmbeddedKafka</code>) but now it complains about type info not found:</p>
  
  <pre><code>o.s.k.listener.LoggingErrorHandler - Error while processing: null org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition checkout_order_paid-0 at offset 0. If needed, please seek past the record to continue consumption.
  Caused by: java.lang.IllegalStateException: No type information in headers and no default type provided
      at org.springframework.util.Assert.state(Assert.java:73)
      at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:326)
      at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1041)
      at org.apache.kafka.clients.consumer.internals.Fetcher.access$3300(Fetcher.java:110)
      at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.fetchRecords(Fetcher.java:1223)
      at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.access$1400(Fetcher.java:1072)
      at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:562)
      at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:523)
      at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1230)
      at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1187)
      at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1154)
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:719)
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:676)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
      at java.util.concurrent.FutureTask.run(FutureTask.java)
      at java.lang.Thread.run(Thread.java:748)
  </code></pre>
  
  <p>I am aware of the documentation, but I don't know how to do it properly:</p>
  
  <p><a href="https://docs.spring.io/spring-kafka/reference/html/#spring-messaging-message-conversion" rel="nofollow noreferrer">https://docs.spring.io/spring-kafka/reference/html/#spring-messaging-message-conversion</a></p>