title: How to deserialize json list with object in my kafka consumer?
body: <p>My Kafka Producer is sending a list of objects in Json format. </p>
  
  <p>I'm trying to figure out how to make my consumer deserialize the list. I'm able to receive a single object and read it but when i'm change the code to type List i'm getting the following error:</p>
  
  <pre><code>Error:(32, 47) java: incompatible types: cannot infer type arguments for org.springframework.kafka.core.DefaultKafkaConsumerFactory&lt;&gt;
      reason: inference variable V has incompatible equality constraints java.util.List&lt;nl.domain.X&gt;,nl.domain.X
  </code></pre>
  
  <p><strong>EDIT</strong> </p>
  
  <p>This error has been solved by adding TypeReference to the JsonDeserilizer.</p>
  
  <p><strong>Current problem</strong>:</p>
  
  <p>When consuming the message, it isn't in the type i defined (which is List&lt; X > ) but returning a <strong>LinkedHashMap</strong></p>
  
  <p>This is the <strong>Consumer</strong> configuration:</p>
  
  <pre><code>@EnableKafka
  @Configuration
  public class KafkaConfiguration {
  
      @Bean
      public ConsumerFactory&lt;String, List&lt;X&gt;&gt; xConsumerFactory() {
          Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
  
          config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
          config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_json");
          config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
          return new DefaultKafkaConsumerFactory&lt;&gt;(config, new StringDeserializer(),
                  new JsonDeserializer&lt;&gt;(new TypeReference&lt;List&lt;X&gt;&gt;() {}));
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, List&lt;X&gt;&gt; xKafkaListenerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, List&lt;X&gt;&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(xConsumerFactory());
          return factory;
      }
  }
  </code></pre>
  
  <p>This is the <strong>Consumer</strong>:</p>
  
  <pre><code>@Service
  public class KafkaConsumer {
      @KafkaListener(topics = "test", groupId = "group_json", containerFactory = "xKafkaListenerFactory")
      public void consume(List&lt;X&gt; x) {
          // In reality it is consuming LinkedHashMap which isn't what i want
          x.forEach(i -&gt;
                  System.out.println("Consumed message: " + i.getName()));
      }
  }
  </code></pre>
  
  <p>This is the <strong>Producer</strong> configuration:</p>
  
  <pre><code>@Configuration
  public class KafkaConfiguration {
      @Bean
      public ProducerFactory&lt;String, List&lt;X&gt;&gt; producerFactory() {
          Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
          config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
          config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
  
          return new DefaultKafkaProducerFactory&lt;&gt;(config);
      }
  
      @Bean
      public KafkaTemplate&lt;String, List&lt;X&gt;&gt; kafkaTemplate() {
          return new KafkaTemplate&lt;&gt;(producerFactory());
      }
  }
  </code></pre>
  
  <p>This is the <strong>producer</strong>: </p>
  
  <pre><code>@Component
  public class KafkaProducer {
  
      private final static String TOPIC = "test";
      private final KafkaTemplate&lt;String, List&lt;X&gt;&gt; kafkaTemplate;
  
      public KafkaProducer(KafkaTemplate&lt;String, List&lt;X&gt;&gt; kafkaTemplate) {
          this.kafkaTemplate = kafkaTemplate;
      }
  
      @Override
      public void publishMessage(List&lt;X&gt; x) {
          // Created 3 instances of X for current example
          List&lt;X&gt; list = new ArrayList();
          list.add(new X("Apple"));
          list.add(new X("Beach"));
          list.add(new X("Money"));
          ListenableFuture&lt;SendResult&lt;String, List&lt;X&gt;&gt;&gt; listenableFuture = kafkaTemplate.send(TOPIC, list);
      }
  }
  </code></pre>
  
  <p><strong>Summary</strong>:</p>
  
  <p>The producer seems to work fine. I'm getting the following error in my consumer when sending a message like this. It can't cast LinkedHashMap into List.</p>
  
  <pre><code>org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void nl.infrastructure.input.message.kafka.consumer.KafkaConsumer.consume(java.util.List&lt;nl.domain.X&gt;)' threw exception; nested exception is java.lang.ClassCastException: class java.util.LinkedHashMap cannot be cast to class nl.domain.X (java.util.LinkedHashMap is in module java.base of loader 'bootstrap'; nl.domain.X is in unnamed module of loader 'app'); nested exception is java.lang.ClassCastException: class java.util.LinkedHashMap cannot be cast to class nl.domain.X (java.util.LinkedHashMap is in module java.base of loader 'bootstrap'; nl.domain.X is in unnamed module of loader 'app')
  </code></pre>