Using spring Kafka to product and listen to the request, The key value in inside the ""

Producer code

public record ProductProducer(ReplyingKafkaTemplate<String, Object, Object> _replyTemplate,
                              ) implements IProductProducer {
     public ProductViewModel Update(ProductViewModel product, String id) throws InterruptedException, ExecutionException, TimeoutException {
            RequestReplyFuture<String, Object, Object> future =
                    this._replyTemplate.sendAndReceive(new ProducerRecord<>(ProductTopicConstants.UPDATE_PRODUCT,0, id,product));
            LOG.info(future.getSendFuture().get(kafkaConstants.kafkaTimeout, TimeUnit.SECONDS).getRecordMetadata().toString());
            Product productDb = (Product) future.get(kafkaConstants.kafkaTimeout, TimeUnit.SECONDS).value();
            return new ProductViewModel();
        }}

Listener

@KafkaListener(id = ProductTopicConstants.UPDATE_PRODUCT, topics = ProductTopicConstants.UPDATE_PRODUCT,
            containerFactory = "addUpdateProductContainerFactory")
    @SendTo
    public Object UpdateProduct(ConsumerRecord<String, ProductViewModel> productViewModel) {
        String id = productViewModel.key();
        logger.info("Listening to update product with id :", id);
        return new Product();
    }

The key value is inside "" as shown below

consumer factory

 @Bean
    public ConsumerFactory<String, String> consumerFactoryGetDeleteProduct() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(),
                new StringDeserializer(),
                new JsonDeserializer<>(String.class));
    }

I have the below configuration

producer:
      key-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      key-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer




@Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapAddress.getKafka().getBootstrapAddress());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }