title: Deserializing different JSON payload from same Kafka topic with Spring Kafka
body: <p>I'm trying to deserialize different JSON payloads from the same Kafka topic. The other questions asked here, guided me to a first attempt, but I was not able to get it running.</p>
  
  <p>As Gary mentioned (<a href="https://stackoverflow.com/a/51695630">here</a>) there is some hint (<em>JsonSerializer.ADD_TYPE_INFO_HEADERS</em>), but when I send and receive both messages I get an exception.</p>
  
  <pre class="lang-java prettyprint-override"><code>org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
  Endpoint handler details:
  Method [public void com.foo.message.ConsumerImpl.consumeSelf(java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.util.Map&lt;java.lang.String, java.lang.Object&gt;,com.foo.message.KafkaMessage,org.apache.kafka.clients.consumer.ConsumerRecord&lt;java.lang.String, java.lang.Object&gt;)]
  Bean [com.foo.message.ConsumerImpl@6df2a206]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [com.foo.message.KafkaMessageWithAdditionalField] to [com.foo.message.KafkaMessage] for GenericMessage [payload=com.foo.message.KafkaMessageWithAdditionalField@4e3168f7, headers={kafka_offset=22, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@c0e2fcf, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=fromBar, kafka_receivedTimestamp=1548310583481}], failedMessage=GenericMessage [payload=com.foo.message.KafkaMessageWithAdditionalField@4e3168f7, headers={kafka_offset=22, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@c0e2fcf, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=fromBar, kafka_receivedTimestamp=1548310583481}]
      at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:292) ~[spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:79) ~[spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:50) ~[spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1207) [spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
  </code></pre>
  
  <p>...</p>
  
  <p>The LoggingErrorHandler mentions already a (the correct) value in ConsumerRecord.</p>
  
  <pre class="lang-java prettyprint-override"><code>2019-01-24 07:16:27.630 ERROR 27204 --- [ntainer#2-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: ConsumerRecord(topic = fromBar, partition = 0, offset = 22, CreateTime = 1548310583481, serialized key size = -1, serialized value size = 196, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = com.foo.bar.message.KafkaMessageWithAdditionalField@4e3168f7)
  </code></pre>
  
  <p>First my config:</p>
  
  <pre class="lang-java prettyprint-override"><code>@EnableKafka
  @Configuration
  public class KafkaConsumerConfig {
      @Value("${spring.kafka.bootstrap-servers}")
      private String bootstrapServers;
  
      @Bean
      public ConsumerFactory&lt;String, KafkaMessage&gt; consumerFactoryMessage()
      {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);           
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
          return new DefaultKafkaConsumerFactory&lt;&gt;(props, new StringDeserializer(),
                  new JsonDeserializer&lt;&gt;(KafkaMessage.class));
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, KafkaMessage&gt; kafkaListenerMessageContainerFactory()
      {
          ConcurrentKafkaListenerContainerFactory&lt;String, KafkaMessage&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactoryMessage());
          return factory;
      }
  
      @Bean
      public ConsumerFactory&lt;String, KafkaMessageWithAdditionalField&gt; consumerFactoryMessageWithAdditionalField()
      {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
          return new DefaultKafkaConsumerFactory&lt;&gt;(props, new StringDeserializer(),
                  new JsonDeserializer&lt;&gt;(KafkaMessageWithAdditionalField.class));
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, KafkaMessageWithAdditionalField&gt; kafkaListenerMessageWithAdditionalFieldContainerFactory()
      {
          ConcurrentKafkaListenerContainerFactory&lt;String, KafkaMessageWithAdditionalField&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactoryMessageWithAdditionalField());
          return factory;
      }
  }
  </code></pre>
  
  <p>Here are the listeners:</p>
  
  <pre class="lang-java prettyprint-override"><code>    @KafkaListener(topicPartitions = @TopicPartition(partitions = "0", topic = "${foo.kafka.topic-springBoot}"), containerFactory = "kafkaListenerMessageContainerFactory")
      public void consumeSelf(@Headers Map&lt;String, Object&gt; map, KafkaMessage message, ConsumerRecord&lt;String, Object&gt; cr)
      {
          log.info("message received %s", message);
      }
  
      @KafkaListener(topicPartitions = @TopicPartition(partitions = "0", topic = "${foo.kafka.topic-springBoot}"), containerFactory = "kafkaListenerMessageWithAdditionalFieldContainerFactory")
      public void consumeSelfAdd(@Headers Map&lt;String, Object&gt; map, KafkaMessageWithAdditionalField message, ConsumerRecord&lt;String, Object&gt; cr)
      {
          log.info("messageKafkaMessageWithAdditionalField received %s", message);
      }
  </code></pre>