Spring Boot Kafka生产者消费者配置问题：对象反序列化失败

核心问题定位 #

你遇到的RecordDeserializationException确实出在消费端反序列化环节，结合你的配置来看，主要集中在生产者与消费者的JSON序列化配置一致性、实体类兼容性、消息内容合法性这几个方向，以下是具体解决步骤：

1. 对齐生产者与消费者的JSON序列化配置 #

你在生产者中使用JsonSerializer.noTypeInfo()（不发送类型头），消费者对应使用JsonDeserializer.ignoreTypeHeaders()是正确的，但要确保两者使用完全一致的ObjectMapper配置，避免默认行为差异导致解析失败：

• 先定义统一的ObjectMapper Bean：

  @Bean
  public ObjectMapper kafkaObjectMapper() {
      ObjectMapper mapper = new ObjectMapper();
      // 忽略未知字段，避免实体类字段与JSON结构不匹配时触发异常
      mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
      // 按需添加其他配置，比如日期格式化、枚举序列化规则等
      return mapper;
  }
  

• 修改生产者配置，传入统一的ObjectMapper：

  return new DefaultKafkaProducerFactory<>(configProps,
          new StringSerializer(),
          new JsonSerializer<Deposit>(kafkaObjectMapper()).noTypeInfo());
  

• 修改消费者配置，传入统一的ObjectMapper并指定信任包：

  return new DefaultKafkaConsumerFactory<>(props, 
          new StringDeserializer(), 
          new JsonDeserializer<>(Deposit.class, kafkaObjectMapper())
              .ignoreTypeHeaders()
              .trustedPackages("com.yourpackage.model")); // 替换为Deposit类所在的包，或用"*"信任所有包
  

2. 检查Deposit实体类的序列化兼容性 #

Jackson反序列化对实体类有几个硬性要求：

• 必须有无参构造函数（用Lombok的@Data注解会自动生成，手动编写时需显式定义）
• 所有需要序列化的字段必须有对应的getter/setter方法（或用@JsonProperty注解标记字段）
• 如果实体类包含复杂类型（自定义对象、集合等），要确保这些类型同样满足序列化要求

3. 验证消息内容的合法性 #

先确认生产者发送的消息是否为合法JSON格式：

• 在生产者发送前，打印序列化后的JSON字符串：

  Deposit deposit = new Deposit();
  // 为字段赋值
  String json = kafkaObjectMapper().writeValueAsString(deposit);
  System.out.println("发送的JSON内容：" + json);
  

• 或者用Kafka命令行工具直接查看消息内容：

  kafka-console-consumer.sh --bootstrap-server your-bootstrap-server:9092 --topic exchange-service-2 --from-beginning --property print.value=true --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
  

如果输出内容不是合法JSON，说明生产者的序列化配置存在问题，需要调整。

4. 补全消费者的必要配置 #

在消费者的props中显式指定反序列化器类，避免隐式配置冲突：

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getName());

5. 处理卡住的偏移量 #

如果之前的错误消息导致消费偏移量卡住，可以手动跳过该条记录：

• 使用命令行重置偏移量（示例：跳过offset 1，直接从offset 2开始消费）：

  kafka-consumer-groups.sh --bootstrap-server your-bootstrap-server:9092 --group your-group-id --topic exchange-service-2 --reset-offsets --to-offset 2 --execute
  

• 或者在@KafkaListener中添加错误处理器，自动跳过失败的消息：

  @KafkaListener(topics = "exchange-service-2", containerFactory = "kafkaListenerContainerFactoryDeposit", errorHandler = "kafkaErrorHandler")
  public void consumeDeposit(Deposit deposit) {
      // 消费逻辑
  }
  
  @Bean
  public KafkaErrorHandler kafkaErrorHandler() {
      return (record, exception) -> {
          System.err.println("消费失败，跳过消息offset：" + record.offset());
          // 可添加日志记录或其他补偿逻辑
      };
  }
  

内容的提问来源于stack exchange，提问作者Nikotin