如何将GenericData转为JsonNode？Kafka消费Avro消息遇类型转换异常

核心问题 #

你在@KafkaListener方法里把ConsumerRecord的泛型类型指定成了GenericData，但实际Kafka客户端返回的消息值是GenericData.Record（Avro通用记录的具体实现类），这就导致了类型转换异常。

修正步骤 #

1. 修正泛型类型
   把方法参数的泛型改成GenericData.Record，匹配实际返回的类型：

   @KafkaListener(topics = "#{'${kafka.topic.name}'}", properties = {"specific.avro.reader=false"})
   public void consume(ConsumerRecord<String, GenericData.Record> record) 
   

2. 优化Avro转JsonNode的方式
   直接调用toString()可能会出现格式不兼容问题，推荐两种更可靠的转换方式：

   • 方式一：使用Avro原生工具类

     try {
         GenericData.Record avroRecord = record.value();
         StringWriter writer = new StringWriter();
         JsonEncoder encoder = EncoderFactory.get().jsonEncoder(avroRecord.getSchema(), writer);
         DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroRecord.getSchema());
         datumWriter.write(avroRecord, encoder);
         encoder.flush();
         JsonNode recordNode = mapper.readTree(writer.toString());
     } catch (IOException ex) {
         throw new IllegalArgumentException("Failed to parse consumer record " + record.key(), ex);
     }
     

   • 方式二：使用Jackson Avro模块（更简洁）
     先引入Jackson Avro依赖，然后直接通过ObjectMapper转换：

     // 初始化带有Avro模块的ObjectMapper
     ObjectMapper mapper = new ObjectMapper().registerModule(new AvroModule());
     // 直接将Avro Record转为JsonNode
     JsonNode recordNode = mapper.valueToTree(record.value());
     

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