Search code examples
javaspringspring-bootspring-kafka

Failed to convert from JSON;Unexpected character ('f' (code 102)): Expected space separating root-level values at (String)"5f19a7e99933db43cb23e83d"

   @KafkaListener(id = ProductTopicConstants.GET_PRODUCT, topics = ProductTopicConstants.GET_PRODUCT)
    @SendTo
    public Product GetProduct(String id) {
        return _productRepository.findByid(id);
    }

Kafka Configuration

    @Configuration
        public class KafkaConfiguration {

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

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


            @Bean
            public ReplyingKafkaTemplate<String, Object, Object> replyer(ProducerFactory<String, Object> pf,
                                                                                     ConcurrentKafkaListenerContainerFactory<String, Object> containerFactory) {
        
                containerFactory.setReplyTemplate(kafkaTemplate(pf));
                ConcurrentMessageListenerContainer<String, Object> container = replyContainer(containerFactory);
                ReplyingKafkaTemplate<String, Object, Object> replyer = new ReplyingKafkaTemplate<>(pf, container);
                return replyer;
            }
        
            @Bean
            public ConcurrentMessageListenerContainer<String, Object> replyContainer(
                    ConcurrentKafkaListenerContainerFactory<String, Object> containerFactory) {
        
                ConcurrentMessageListenerContainer<String, Object> container =
                        containerFactory.createContainer(ProductTopicConstants.LISTNER_CONTAINER);
                container.getContainerProperties().setGroupId(ProductTopicConstants.LISTNER_CONTAINER);
                containerFactory.setMessageConverter(new JsonMessageConverter());
                container.setBatchErrorHandler(new BatchLoggingErrorHandler());
                return container;
            }
        
            @Bean
            public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> pf) {
                return new KafkaTemplate<>(pf);
            }
        }

Kafka Producer

   @Override
    public ProductViewModel GetProduct(String id) throws InterruptedException, ExecutionException, TimeoutException {
        RequestReplyFuture<String, Object, Object> future =
                this._replyTemplate.sendAndReceive(new ProducerRecord<>(ProductTopicConstants.GET_PRODUCT, 0, null,id));
        LOG.info(future.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());
        var products = (Product)future.get(10, TimeUnit.SECONDS).value();
        var mappedProducts = mapper.convertValue(products, new TypeReference<Product>() { });
        return new ProductViewModel(mappedProducts.getId(), mappedProducts.getName(), mappedProducts.getPrice(), mappedProducts.getDescription());
    }

Kafka Configuration

spring:
  profiles:
    active: dev
  kafka:
    bootstrapAddress: localhost:9092
    producer:
      key-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest
      properties:
        spring:
          json:
            trusted:
              packages: '*'

Topic

@Bean
    public static NewTopic GetProduct() {
        return new NewTopic(ProductTopicConstants.GET_PRODUCT, 1, (short) 1);
    }

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

Error

org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:157) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:103) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1887) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1792) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1719) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1617) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1348) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1064) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:972) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
Caused by: org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1902) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 11 common frames omitted
Caused by: org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.support.converter.JsonMessageConverter.extractAndConvertValue(JsonMessageConverter.java:117) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.support.converter.MessagingMessageConverter.toMessage(MessagingMessageConverter.java:123) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.toMessagingMessage(MessagingMessageListenerAdapter.java:309) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:77) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1854) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1836) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1779) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 9 common frames omitted
Caused by: com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1851) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:707) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:632) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportMissingRootWS(ParserMinimalBase.java:679) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._verifyRootSpace(ReaderBasedJsonParser.java:1703) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._parsePosNumber(ReaderBasedJsonParser.java:1346) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:773) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4620) ~[jackson-databind-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4469) ~[jackson-databind-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3434) ~[jackson-databind-2.11.0.jar:2.11.0]
    at org.springframework.kafka.support.converter.JsonMessageConverter.extractAndConvertValue(JsonMessageConverter.java:114) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 16 common frames omitted

2020-07-26 21:52:47.026  INFO 3997 --- [etProduct-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-GetProduct-4, groupId=GetProduct] Seeking to offset 25 for partition GetProduct-0
2020-07-26 21:52:47.029 ERROR 3997 --- [etProduct-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:157) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:103) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1887) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1792) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1719) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1617) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1348) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1064) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:972) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
Caused by: org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1902) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 11 common frames omitted
Caused by: org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.support.converter.JsonMessageConverter.extractAndConvertValue(JsonMessageConverter.java:117) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.support.converter.MessagingMessageConverter.toMessage(MessagingMessageConverter.java:123) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.toMessagingMessage(MessagingMessageListenerAdapter.java:309) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:77) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1854) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1836) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1779) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 9 common frames omitted
Caused by: com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1851) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:707) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:632) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportMissingRootWS(ParserMinimalBase.java:679) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._verifyRootSpace(ReaderBasedJsonParser.java:1703) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._parsePosNumber(ReaderBasedJsonParser.java:1346) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:773) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4620) ~[jackson-databind-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4469) ~[jackson-databind-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3434) ~[jackson-databind-2.11.0.jar:2.11.0]
    at org.springframework.kafka.support.converter.JsonMessageConverter.extractAndConvertValue(JsonMessageConverter.java:114) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 16 common frames omitted

2020-07-26 21:52:47.534  INFO 3997 --- [etProduct-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-GetProduct-4, groupId=GetProduct] Seeking to offset 25 for partition GetProduct-0
2020-07-26 21:52:47.537 ERROR 3997 --- [etProduct-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:157) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:103) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1887) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1792) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1719) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1617) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1348) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1064) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:972) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
Caused by: org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1902) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 11 common frames omitted
Caused by: org.springframework.kafka.support.converter.ConversionException: Failed to convert from JSON; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at org.springframework.kafka.support.converter.JsonMessageConverter.extractAndConvertValue(JsonMessageConverter.java:117) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.support.converter.MessagingMessageConverter.toMessage(MessagingMessageConverter.java:123) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.toMessagingMessage(MessagingMessageListenerAdapter.java:309) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:77) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1854) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1836) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1779) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 9 common frames omitted
Caused by: com.fasterxml.jackson.core.JsonParseException: Unexpected character ('f' (code 102)): Expected space separating root-level values
 at [Source: (String)"5f19a7e99933db43cb23e83d"; line: 1, column: 3]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1851) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:707) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:632) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportMissingRootWS(ParserMinimalBase.java:679) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._verifyRootSpace(ReaderBasedJsonParser.java:1703) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._parsePosNumber(ReaderBasedJsonParser.java:1346) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:773) ~[jackson-core-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4620) ~[jackson-databind-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4469) ~[jackson-databind-2.11.0.jar:2.11.0]
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3434) ~[jackson-databind-2.11.0.jar:2.11.0]
    at org.springframework.kafka.support.converter.JsonMessageConverter.extractAndConvertValue(JsonMessageConverter.java:114) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    ... 16 common frames omitted

2020-07-26 21:52:48.049  INFO 3997 --- [etProduct-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-GetProduct-4, groupId=GetProduct] Seeking to offset 25 for partition GetProduct-0
2020-07-26 21:52:48.052 ERROR 3997 --- [etProduct-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

Output in Consumer console

enter image description here

All the combination I have tried but none of them works

Solution

  • You need to show the consumer side configuration; you are doing double JSON conversion JsonMessageConverter is being applied, which is unnecessary since you have already converted the JSON to a String in the JsonDeserializer. You either need to replace the deserializer with a simple StringDeserializer or ByteArrayDeserializer so the converter has something to convert.

    EDIT

    So, the problem is that your ProductListener has a bean

    @Bean
public RecordMessageConverter converter() {
    return new StringJsonMessageConverter();
}

    and Boot is auto-configuring it into the listener container factory and you have a JsonDeserializer so, as I said, you are doing double deserialization. I was able to reproduce it by adding that message converter bean into your stripped down app.

    You can either use a different factory for that listener, or override the deserializer on the listener itself...

    @KafkaListener(id = ProductTopicConstants.GET_PRODUCT, topics = ProductTopicConstants.GET_PRODUCT,
        clientIdPrefix = "getProd", 
        properties = ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
            + "=org.apache.kafka.common.serialization.StringDeserializer")