我也面临过这个问题，但是上面的解决方案对我不起作用。不过，关键在于将Kafka消费品工厂配置如下：

props.put(JsonDeserializer.TRUSTED_PACKAGES, "your.package.name");

请参阅文档。
从版本2.1开始，类型信息可以在记录头中传递，允许处理多种类型。此外，可以使用kafka属性配置序列化程序/反序列化程序。
jsonserializer.add\ type\ info\ headers（默认为true）；设置为false可禁用jsonserializer上的此功能（设置addtypeinfo属性）。
jsondeserializer.key\默认\类型；如果不存在标头信息，则用于键反序列化的回退类型。
jsondeserializer.value\默认\类型；如果不存在标头信息，用于反序列化值的回退类型。
jsondeserializer.trusted\u包（默认java.util、java.lang）；允许反序列化的包模式的逗号分隔列表；*意味着反序列化所有。
默认情况下，序列化程序将向标头添加类型信息。
请参阅引导文档。
类似地，您可以禁用jsonserializer在标头中发送类型信息的默认行为：
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer spring.kafka.producer.properties.spring.json.add.type.headers=false 或者可以将类型Map添加到入站消息转换器，以将源类型Map到目标类型。
编辑
说到这里，你用的是什么版本？

jsonDeserializer.addTrustedPackages("*");

为spring-kafka-2.2.8解决了我的问题。
添加到 application.properties :

spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*

重要提示：
如果您分别为kafkaconsumer和kafkaproducer提供了序列化程序和反序列化程序示例，则它们没有任何效果。
参考文献：
[1] https://docs.spring.io/spring-kafka/reference/html/#json-塞德
[2] https://github.com/spring-projects/spring-kafka/issues/535

生产者配置类

import com.kafka.producer.models.Container;    
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {

@Bean
public ProducerFactory<String, Container> producerFactory(){

    Map<String, Object> config = new HashMap<>();

config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

    return new DefaultKafkaProducerFactory(config);
}

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

注意：容器是要在kafka主题中发布的自定义对象。

生产者阶级

import com.kafka.producer.models.Container;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class Producer {

private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
private static final String TOPIC = "final-topic";

@Autowired
private KafkaTemplate<String, Container> kafkaTemplate;

public void sendUserMessage(Container msg) {
    LOGGER.info(String.format("\n ===== Producing message in JSON ===== \n"+msg));
    Message<Container> message = MessageBuilder
            .withPayload(msg)
            .setHeader(KafkaHeaders.TOPIC, TOPIC)
            .build();
    this.kafkaTemplate.send(message);
}
}

生产商控制器

import com.kafka.producer.models.Container;
import com.kafka.producer.services.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/message")
public class MessageController {

@Autowired
private Producer producer;

@PostMapping(value = "/publish")
public String sendMessageToKafkaTopic(@RequestBody Container containerMsg) {
    this.producer.sendUserMessage(containerMsg);
    return "Successfully Published !!";
}
}

注意：container类型的消息将作为json消息发布到kafka topic name:final topic。

--消费者应用程序--

配置类

import com.kafka.consumer.models.Container;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConsumerOneConfig {

@Bean
public ConsumerFactory<String, Container> consumerFactory(){
    JsonDeserializer<Container> deserializer = new JsonDeserializer<>(Container.class);
    deserializer.setRemoveTypeHeaders(false);
    deserializer.addTrustedPackages("*");
    deserializer.setUseTypeMapperForKey(true);

    Map<String, Object> config = new HashMap<>();

    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_one");
    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deserializer);

    return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), deserializer);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Container> kafkaListenerContainerFactory(){
    ConcurrentKafkaListenerContainerFactory<String, Container> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;

}
}

注意：在这里您可以看到，我们必须使用自定义的jsondeserializer来使用来自最终主题（主题名称）的容器对象类型json消息，而不是使用默认的jsondeserializer（）。

消费者服务

import com.kafka.consumer.models.Container;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import java.io.IOException;

@Service
public class ConsumerOne {

private final Logger LOGGER = LoggerFactory.getLogger(ConsumerOne.class);

@KafkaListener(topics = "final-topic", groupId = "group_one", containerFactory = "kafkaListenerContainerFactory")
public void consumeUserMessage(@Payload Container msg, @Headers MessageHeaders headers) throws IOException {
    System.out.println("received data in Consumer One ="+ msg.getMessageTypes());
}
}

对于这一个，有两种方法，要么在反序列化程序中，要么在application.yml中。

在反序列化程序中

在反序列化程序中，在 DefaultKafkaConsumerFactory （创建消费工厂）。假设你想做一个 ConsumerFactory<String, Foo> 与 Foo 成为Kafka信息中的模特/pojo。
你需要 addTrustedPackages 从 JsonDeserializer 我在kotlin中有一个例子，但在java中几乎是相同的语法：

val deserializer = JsonDeserializer<Foo>()
 deserializer.addTrustedPackages("com.example.entity.Foo") // Adding Foo to our trusted packages

 val consumerFactory = DefaultKafkaConsumerFactory(
      consumerConfigs(),  // your consumer config
      StringDeserializer(), 
      deserializer // Using our newly created deserializer
 )

或者在你的应用程序中

在application.yml文件中，遵循spring kafka指令。我们将com.example.entity.foo包中的foo类添加到受信任的存储中，使用：

spring:
  kafka:
    consumer:
      properties:
        spring.json.trusted.packages: "com.example.entity.Foo"

与 spring.json.trusted.packages 接受包数组。可以指定类包，也可以使用 * 任何包裹。那样的话你就不用通过考试了 deserializer 在 DefaultKafkaConsumerFactory() 仅在使用者配置中。

应该提到两个关键点。
有两个独立的项目为生产者和消费者。
然后发送消息（值）是一种对象类型，而不是原始类型。
问题是产生消息对象在使用者端不可用，因为这是两个独立的项目。
两个克服这个问题请遵循下面提到的步骤在 Spring 启动生产者和消费者的应用程序。
----生产商应用程序

我的SpringKafka版本是2.2.11，我也有这个错误。
我得到这个错误是因为我在同一个kafta主题中用不同的配置配置了两个消费者。其中一个有consumerfactory<string，ordedto>，另一个有consumerfactory<string，string>。
我解决了更改一个消费者配置的错误，因为这是错误的。
只需检查主题的消费者