Quarkus Kafka消费者如何反序列化含嵌套对象的复杂Person类？

一、修正Kafka消费者配置 #

你的配置存在明显矛盾：生产者用ObjectMapperSerializer序列化JSON格式数据，消费者却配置了StringDeserializer，这会导致消费到的是JSON字符串而非直接可反序列化的Person对象。需要将消费者的反序列化器改为对应Jackson实现：

修改配置文件内容：

%prod.mp.messaging.incoming.topic_test.connector=smallrye-kafka
%prod.mp.messaging.incoming.topic_test.topic=test-topic
%prod.mp.messaging.incoming.topic_test.value.serializer=io.quarkus.kafka.client.serialization.ObjectMapperSerializer
# 替换成Jackson反序列化器
%prod.mp.messaging.incoming.topic_test.value.deserializer=io.quarkus.kafka.client.serialization.ObjectMapperDeserializer
# 指定反序列化的目标类全路径
%prod.mp.messaging.incoming.topic_test.value.deserializer.class=your.package.path.Person

注意：value.deserializer.class必须填写Person类的实际全限定包路径，让Jackson明确知道要将JSON转成哪个对象。

二、解决Jackson构造函数问题 #

报错提示“no Creators, like default constructor, exist”，核心原因是Jackson无法找到合适的构造函数实例化Person对象。虽然你使用了Lombok的@Data，但@Data默认包含@RequiredArgsConstructor（仅生成带必填字段的构造函数），不会自动生成无参构造函数，而Jackson默认依赖无参构造来创建对象。

解决方法二选一：

1. 给所有实体类添加无参构造（推荐）
   用Lombok的@NoArgsConstructor配合@Data使用，所有嵌套实体类也需要添加：

   @Data
   @NoArgsConstructor
   public class Person {
       private Hobbies hobbies;
       private List<Relatives> relatives;
   }
   
   @Data
   @NoArgsConstructor
   public class Hobbies {
       private Sports sports;
       private String principal;
   }
   
   @Data
   @NoArgsConstructor
   public class Parents {
       private Children kids;
       private String mom;
   }
   

   注意：Sports、Relatives、Children等所有嵌套类都需要添加@Data + @NoArgsConstructor，避免嵌套对象反序列化失败。

2. 配置Jackson使用带参构造
   若不想用无参构造，可给带参构造的参数添加@JsonProperty注解，让Jackson识别参数与字段的对应关系：

   @Data
   @AllArgsConstructor
   public class Person {
       @JsonProperty
       private Hobbies hobbies;
       @JsonProperty
       private List<Relatives> relatives;
   }
   

   这种方式需要所有嵌套类同步处理，相对繁琐，仅适合必须使用带参构造的场景。

三、处理滞留消息 #

之前的错误导致消息滞留在分区，可按以下方式处理：

• 丢弃错误消息：通过Kafka命令行工具重置消费者组偏移量到最新位置，跳过滞留消息：

  kafka-consumer-groups.sh --bootstrap-server your-kafka-host:9092 --group your-consumer-group-id --reset-offsets --to-latest --topic test-topic --execute
  

  替换your-kafka-host和your-consumer-group-id为实际值，消费者组ID可在配置中通过mp.messaging.incoming.topic_test.group.id指定，未指定则Quarkus会生成默认值。
• 保留消息并重试：修复代码和配置后重启应用，应用会自动尝试重新消费滞留的消息。

额外检查点 #

• 确认所有嵌套实体类的结构完整，字段与消息JSON的键名一致（区分大小写）。
• 验证Kafka中消息的JSON格式是否符合Person类的结构，避免因消息本身格式错误导致反序列化失败。

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