title: How to consume and save a List of custom types in Spring Boot using Apache Kafka?
body: <p>I am writing a Kafka producer and consumer to send my custom data and save after consuming. I did all the required Kafka configurations, but I get a casting exception. </p>
  
  <p>These are the codes:</p>
  
  <p>Consumer:</p>
  
  <pre><code>public class Consumer {
      private final Logger logger = LoggerFactory.getLogger(Consumer.class);
  
      @Autowired
      private RecordService service;
  
      @KafkaListener(topics = "data", groupId = "group_id", containerFactory = "kafkaListenerContainerFactory")
      public void consume(List&lt;Record&gt; message) {
          logger.info(String.format("$$ -&gt; Consumed Message -&gt; %s", "message"));
  
          System.out.println(message);
          service.saveAll(message);
      }
  }
  </code></pre>
  
  <p>Producer:</p>
  
  <pre><code>public class Producer {
      private static final Logger logger = LoggerFactory.getLogger(Producer.class);
      private static final String TOPIC = "data";
      @Autowired
      private KafkaTemplate&lt;String, List&lt;Record&gt;&gt; kafkaTemplate;
  
      public void sendMessage(List&lt;Record&gt; list) {
          logger.info(String.format("$$ -&gt; Producing message --&gt; %s", "message"));
          // Message&lt;List&lt;Record&gt;&gt; message =
          // MessageBuilder.withPayload(list).setHeader(KafkaHeaders.TOPIC,
          // TOPIC).build();
  
          this.kafkaTemplate.send(TOPIC, list);
      }
  }
  </code></pre>
  
  <p>KafkaConfig for consumer</p>
  
  <pre><code>@Configuration
  public class KafkaConfiguration {
  
      @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, JsonDeserializer.class);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "json");
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          return props;
      }
  
      @Bean
      public ConsumerFactory&lt;String, List&lt;Record&gt;&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs(), new StringDeserializer(),
                  new JsonDeserializer&lt;&gt;(List.class));
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, List&lt;Record&gt;&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, List&lt;Record&gt;&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  }
  </code></pre>
  
  <p>KafkaConfig for producer:</p>
  
  <pre><code>public class KafkaConfiguration {
  
      @Bean
      public Map&lt;String, Object&gt; producerConfigs() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
          return props;
      }
  
      @Bean
      public ProducerFactory&lt;String, List&lt;Record&gt;&gt; producerFactory() {
          return new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());
      }
  
      @Bean
      public KafkaTemplate&lt;String, List&lt;Record&gt;&gt; kafkaTemplate() {
          return new KafkaTemplate&lt;&gt;(producerFactory());
      }
  }
  </code></pre>
  
  <p>The message successfully sent and printed to console, but the type is not what I expect. I send a List, but it is received as a List of LinkedHashMap. There is Stack Trace Error: </p>
  
  <pre><code>2019-08-28 17:41:52.612  INFO 22236 --- [ntainer#0-0-C-1] com.consumerApp.models.Consumer          : $$ -&gt; Consumed Message -&gt; message
  [{region=Central America and the Caribbean, country=Haiti, itemType=Office Supplies, salesChannel=Online, orderPriority=C, orderDate=1293739200000, orderId=485070693, shipDate=1296417600000, unitsSold=2052, unitPrice=651.21, unitCost=524.96, totalRevenue=1336282.92, totalCost=1077217.92, totalProfit=259065.0}, {region=Central America and the Caribbean, country=Nicaragua, itemType=Household, salesChannel=Online, orderPriority=C, orderDate=1445976000000, orderId=573998582, shipDate=1449432000000, unitsSold=7791, unitPrice=668.27, unitCost=502.54, totalRevenue=5206491.57, totalCost=3915289.14, totalProfit=1291202.43}]
  2019-08-28 17:41:52.613 ERROR 22236 --- [ntainer#0-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: ConsumerRecord(topic = data, partition = 0, offset = 16, CreateTime = 1566999712608, serialized key size = -1, serialized value size = 675, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = [{region=Central America and the Caribbean, country=Haiti, itemType=Office Supplies, salesChannel=Online, orderPriority=C, orderDate=1293739200000, orderId=485070693, shipDate=1296417600000, unitsSold=2052, unitPrice=651.21, unitCost=524.96, totalRevenue=1336282.92, totalCost=1077217.92, totalProfit=259065.0}, {region=Central America and the Caribbean, country=Nicaragua, itemType=Household, salesChannel=Online, orderPriority=C, orderDate=1445976000000, orderId=573998582, shipDate=1449432000000, unitsSold=7791, unitPrice=668.27, unitCost=502.54, totalRevenue=5206491.57, totalCost=3915289.14, totalProfit=1291202.43}])
  
  org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.consumerApp.models.Consumer.consume(java.util.List&lt;com.consumerApp.models.Record&gt;)' threw exception; nested exception is java.lang.ClassCastException: java.util.LinkedHashMap cannot be cast to com.consumerApp.models.Record; nested exception is java.lang.ClassCastException: java.util.LinkedHashMap cannot be cast to com.consumerApp.models.Record
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1272) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1261) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1188) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1159) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1099) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:934) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:750) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:699) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_151]
      at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_151]
      at java.lang.Thread.run(Thread.java:748) [na:1.8.0_151]
  Caused by: java.lang.ClassCastException: java.util.LinkedHashMap cannot be cast to com.consumerApp.models.Record
      at com.consumerApp.models.RecordService.saveAll(RecordService.java:15) ~[classes/:na]
      at com.consumerApp.models.Consumer.consume(Consumer.java:23) ~[classes/:na]
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_151]
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_151]
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_151]
      at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_151]
      at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:170) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
      at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:120) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
      at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:48) ~[spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:283) ~[spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:79) ~[spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:50) ~[spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1224) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1217) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1178) [spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
      ... 8 common frames omitted
  </code></pre>
  
  <p>Why isn't it received as a List? Any ideas? Thank you.</p>