Kafka Consumer无法为JsonDeserializer使用905字节的消息

我们正在使用Spring与Kafka的消费者和制作人 . 我们正在生成大小为905字节的消息 . 我们正在序列化消息并尝试为下一个消费者反序列化它 .

Message Payload Class Sample:

{  
        "FILE_LIST":[  
            {
                "KEY1": "Large String Value",
                "KEY2": "Large String Value",
                "Key3": "Large String Value",
                "Key4": "Large String Value"
            }
        ]
}

有效载荷类

import com.fasterxml.jackson.annotation.JsonProperty;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

@NoArgsConstructor
@AllArgsConstructor
public class InnerModel {

    @JsonProperty("KEY1")
    @Getter
    @Setter
    private String key1;

    @JsonProperty("KEY2")
    @Getter
    @Setter
    private String key2;

    @JsonProperty("KEY3")
    @Getter
    @Setter
    private String key3;

    @JsonProperty("KEY4")
    @Getter
    @Setter
    private String key4;
}

CustomModel.java

package com.consumer.model; 


public class CustomModel {
    public CustomModel(List<InnerModel> filesList) {
        super();
        this.filesList = filesList;
    }

    @JsonProperty("FILE_LIST")
    @NonNull
    @Getter
    @Setter
    List<InnerModel> filesList;
}

Consumer Code

import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
@KafkaListener(topics = "customtopic", groupId = "customgroup")
public class CustomModelConsumer {
    @KafkaHandler(isDefault = true)
    private void consumeCustomModel(CustomModel model) {
        System.out.println("Model Consumer");
        System.out.println(model);
    }
}

application.properties

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.compression-type=gzip 
spring.kafka.consumer.properties.spring.json.trusted.packages=com.consumer.model

Consumer works well when we accept message payload in string format, but we are facing problem when we de-serialize payload in consumer to object. 以下错误也是如此

org.apache.kafka.common.errors.SerializationException：在偏移量70处对分区customtopic-0的键/值进行反序列化时出错 . 如果需要，请搜索记录以继续使用 . 引起：org.apache.kafka.common.errors.SerializationException：无法反序列化数据[[123,34,73,78,80,85,84,95,70,73,76,69,95,76， 73,83,84,34,58,91,123,34,79,82,73,71,73,78,65,76,95,67,76,73,80,95,67,85,82， 82,69,78,84,34,58,34,115,51,58,47,47,98,45,97,111,45,112,114,111,100,117,99,116,45， 109,111,99,107,47,55,52,49,48,57,102,98,98,45,54,102,52,101,45,52,48,54,50,45,97， 98,48,102,45,100,49,102,53,98,98,102,55,49,97,56,49,47,99,108,105,112,115,47,48,58， 48,48,58,48,48,46,48,48,48,95,48,58,48,51,58,48,48,46,50,48,48,47,49,48,48， 48,95,111,114,105,103,105,110,97,108,99,108,105,112,47,48,58,48,48,58,48,48,46,48,48， 48,95,48,58,48,51,58,48,48,46,50,48,48,47,48,58,48,48,58,48,48,46,48,48,48， 95,48,58,48,51,58,48,48,46,50,48,48,47,48,58,48,48,58,48,48,46,48,48,48,95， 48,58,48,51,58， 48,48,46,50,48,48,46,109,112,52,34,44,34,79,82,73,71,73,78,65,76,95,67,76,73， 80,95,71,76,79,66,65,76,95,84,73,77,69,83,84,65,77,80,95,67,85,82,82,69,78， 84,34,58,34,115,51,58,47,47,98,45,97,111,45,112,114,111,100,117,99,116,45,109,111,99， 107,47,55,52,49,48,57,102,98,98,45,54,102,52,101,45,52,48,54,50,45,97,98,48,102， 45,100,49,102,53,98,98,102,55,49,97,56,49,47,99,108,105,112,115,47,48,58,48,48,58， 48,48,46,48,48,48,95,48,58,48,51,58,48,48,46,50,48,48,47,49,48,48,48,95,111， 114,105,103,105,110,97,108,99,108,105,112,47,48,58,48,48,58,48,48,46,48,48,48,95,48， 58,48,51,58,48,48,46,50,48,48,47,48,58,48,48,58,48,48,46,48,48,48,95,48,58， 48,51,58,48,48,46,50,48,48,95,116,115,47,48,58,48,48,58,48,48,46,48,48,48,95， 48,58,48,51,58,48,48,46,50,48,48,95,116,115,46,1 06,115,111,110,34,44,34,79,82,73,71,73,78,65,76,95,67,76,73,80,95,78,69,88,84， 34,58,34,115,51,58,47,47,98,45,97,111,45,112,114,111,100,117,99,116,45,109,111,99,107， 47,55,52,49,48,57,102,98,98,45,54,102,52,101,45,52,48,54,50,45,97,98,48,102,45， 100,49,102,53,98,98,102,55,49,97,56,49,47,99,108,105,112,115,47,48,58,48,51,58,48， 48,46,48,48,48,55,48,58,48,54,58,48,48,46,50,48,48,47,49,48,48,48,95,111,114， 105,103,105,110,97,108,99,108,105,112,47,48,58,48,51,58,48,48,46,48,48,48,95,48,58， 48,54,58,48,48,46,50,48,48,47,48,58,48,51,58,48,48,46,48,48,48,95,48,58,48， 54,58,48,48,46,50,48,48,47,48,58,48,51,58,48,48,46,48,48,48,95,48,58,48,54， 58,48,48,46,50,48,48,46,109,112,52,34,44,34,79,82,73,71,73,78,65,76,95,67,76， 73,80,95,71,76,79,66,65,76,95,84,73,77,69,83,84， 65,77,80,95,78,69,88,84,34,58,34,115,51,58,47,47,98,45,97,111,45,112,114,111,100， 117,99,116,45,109,111,99,107,47,55,52,49,48,57,102,98,98,45,54,102,52,101,45,52,48， 54,50,45,97,98,48,102,45,100,49,102,53,98,98,102,55,49,97,56,49,47,99,108,105,112， 115,47,48,58,48,51,58,48,48,46,48,48,48,95,48,58,48,54,58,48,48,46,50,48,48， 47,49,48,48,48,95,111,114,105,103,105,110,97,108,99,108,105,112,47,48,58,48,51,58,48， 48,46,48,48,48,55,48,58,48,54,58,48,48,46,50,48,48,47,48,58,48,51,58,48,48， 46,48,48,48,55,48,58,48,54,58,48,48,46,50,48,48,95,116,115,47,48,58,48,51,58， 48,48,46,48,48,48,55,48,58,48,54,58,48,48,46,50,48,48,95,116,115,46,106,115,111，来自topic [customtopic]的110,34,125,93,125]]来自：com.fasterxml.jackson.databind.exc.InvalidDefinitionException：无法构造com.cons的实例umer.model.CustomModel（没有创建者，就像默认构造一样）：不能在[Source：（byte []）“{”FILE_LIST“：[{”KEY1]中反序列化Object值（没有基于委托或属性的Creator） “：”大字符串值“，”KEY2“：”大字符串值“，”Key3“：”大字符串值“[截断405字节]; line：1，column：2] atcom.fasterxml.jackson.databind.exc.InvalidDefinitionException.from（InvalidDefinitionException.java:67）〜[jackson-databind-2.9.5.jar：2.9.5] at com.fasterxml.jackson.databind.DeserializationContext.reportBadDefinition（DeserializationContext） .java：1451）〜[jackson-databind-2.9.5.jar：2.9.5] at com.fasterxml.jackson.databind.DeserializationContext.handleMissingInstantiator（DeserializationContext.java:1027)~ [jackson-databind-2.9.5 . jar：2.9.5] at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromObjectUsingNonDefault（BeanDeserializerBase.java:1290)~ [jackson-databind-2.9.5.jar：2.9.5] at com.fasterxml.jackson . databind.deser.BeanDeserializer.deserializeFromObject（BeanDeserializer.java:326）〜[jackson-databind-2.9.5.jar：2.9.5] at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize（BeanDeserializer.java:159 ）〜[jackson-databind-2.9.5.jar：2.9.5] at com.fasterxml.jackson.databind.ObjectReader._bindAndClose（ObjectReader.java:1611)~ [jackson-databind-2.9.5.jar：2.9 . 5] at com.fasterxml.jackson.databind.ObjectReader.readValue（ObjectReader.java:1234）〜[jackson-databind-2.9.5.jar：2.9.5] at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize（ orson.apache.k. [kafka-clients-1.0.0.jar：na] at org.apache.kafka.clients.consumer.internals.Fetcher.access $ 2600（Fetcher.java:93）〜[kafka-clients-1.0.0.jar：na ] org.apache.kafka.clients.consumer.internals.Fetcher $ PartitionRecords.fetchRecords（Fetcher.java:1100）〜[kafka-clients-1.0.0.jar：na] at org.apache.kafka.clients.consumer .internals.Fetcher $ PartitionRecords.access $ 1200（Fetcher.java:949)~ [kafka-clients-1.0.0.jar：na] at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords（Fetcher.java） ：570）〜[kafka-clients-1.0.0.jar：na] at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords（Fetcher.java:531）〜[kafka-clients-1.0.0 . jar：na] at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146）〜[kafka-clients-1.0.0.jar：na] at org.apache.kafka.clients.consumer.KafkaConsumer.poll（KafkaConsumer） .java：1103）〜[kafka-clients-1.0.0.jar：na] at org.springframework.kafka.listener.KafkaMessageListenerContainer $ ListenerConsumer.run（KafkaMessageListenerContainer.java:667)~ [spring-kafka-2.1.4 . java.util.concurrent.Executors上的RELEASE.jar：2.1.4.RELEASE] $ java.util.concurrent.FutureTask.run中的RunnableAdapter.call（Executors.java:511）[na：1.8.0_151]（FutureTask.java ：266）[na：1.8.0_151] at java.lang.Thread.run（Thread.java:748)[na:1.8.0_151]