RequestReplyFuture<String, String, List<Product>> not mapped, Instead it mapped to ArrayList<LinkedHashMap>

I am using RequestReplyFuture<String, String, List<Product>> to mapped the response to List<Product>, the result is something like below

    @Service
    public class ProductProducer implements IProductProducer{
      private final ReplyingKafkaTemplate<String, String, List<Product>> _replyTemplate;
        private static final Logger LOG = LoggerFactory.getLogger(ProductProducer.class);
        public ProductProducer(ReplyingKafkaTemplate<String, String, List<Product>> replyTemplate) {
            this._replyTemplate = replyTemplate;
        }
    
     @Override
        public List<ProductViewModel> GetProducts() throws InterruptedException, ExecutionException, TimeoutException {
            RequestReplyFuture<String, String, List<Product>> future =
                    this._replyTemplate.sendAndReceive(new ProducerRecord<>(ProductTopicConstants.GET_PRODUCTS, 0, null, null));
                LOG.info(future.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());
                List<Product> products = future.get(10, TimeUnit.SECONDS).value(); --> Property not mapped to Product
var productViewModels = products.stream().map(item -> new ProductViewModel(item.getId(),item.getName(),item.getPrice(), item.getDescription())).collect(Collectors.toList());
                return productViewModels;
        }
    }

Kafka Configuration

@Configuration
public class KafkaConfiguration {
    @Bean
    public ReplyingKafkaTemplate<String, String, List<Product>> replyer(ProducerFactory<String, String> pf,
                                                                        ConcurrentKafkaListenerContainerFactory<String, List<Product>> containerFactory) {

        containerFactory.setReplyTemplate(kafkaTemplate(pf));
        ConcurrentMessageListenerContainer<String, List<Product>> container = replyContainer(containerFactory);
        ReplyingKafkaTemplate<String, String, List<Product>> replyer = new ReplyingKafkaTemplate<>(pf, container);
        return replyer;
    }

    @Bean
    public ConcurrentMessageListenerContainer<String, List<Product>> replyContainer(
            ConcurrentKafkaListenerContainerFactory<String, List<Product>> containerFactory) {

        ConcurrentMessageListenerContainer<String, List<Product>> container =
                containerFactory.createContainer(ProductTopicConstants.GET_PRODUCTS_CONTAINER);
        container.getContainerProperties().setGroupId(ProductTopicConstants.GET_PRODUCTS_CONTAINER);
        container.setBatchErrorHandler(new BatchLoggingErrorHandler());
        return container;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
        return new KafkaTemplate<>(pf);
    }



   @Bean
    public NewTopic GetProducts() {
        return TopicBuilder.name(ProductTopicConstants.GET_PRODUCTS).partitions(1).replicas(1).build();
    }
    @Bean
    public NewTopic GetProductsContainer() {
        return TopicBuilder.name(ProductTopicConstants.GET_PRODUCTS_CONTAINER).partitions(1).replicas(1).build();
    }
}

Due to type erasure, the type information in the headers makes Jackson think it's List<Object>.

You can use a type function instead, to give Jackson some more information...

spring.kafka.consumer.properties.spring.json.value.type.method=com.example.demo.SomeClass.returnType

public static JavaType returnType(byte[] data, Headers headers) {
	return TypeFactory.defaultInstance()
			.constructCollectionLikeType(List.class, Product.class);
}