Spring Boot Kafka Avro消费者序列化异常求助

从你提供的异常信息和配置来看，核心问题有两个：一是消费者无法找到生产者项目中定义的BookAvro类，二是缺少对序列化异常的处理配置。下面是分步解决方法：

一、解决类找不到的核心问题 #

异常信息中明确提到：Could not find class com.dailycodebuffer.kafka.apachekafkaproducerdemo.BookAvro specified in writer's schema，这是因为生产者生成的BookAvro类包路径和消费者项目中的BookAvro包路径不一致导致的。

解决步骤： #

1. 统一Avro类的包路径
   打开你的Avsc（Avro Schema）文件，修改namespace字段，让生产者和消费者生成的BookAvro类使用完全相同的全限定类名。比如：

   {
     "namespace": "com.your.shared.package", // 替换成你想统一的包路径
     "type": "record",
     "name": "BookAvro",
     "fields": [
       // 你的字段定义
     ]
   }
   

2. 重新生成Avro类
   分别在生产者和消费者项目中，基于修改后的Avsc文件重新生成BookAvro类，确保两边的类全限定名完全一致。

二、配置ErrorHandlingDeserializer处理序列化异常 #

异常中提示：This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer'，需要用ErrorHandlingDeserializer包装实际的反序列化器，让错误处理器能正确处理这类异常。

修改消费者配置： #

@Bean
public ConsumerFactory<String, BookAvro> BookconsumerFactory(){
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
    // 使用ErrorHandlingDeserializer包装键和值的反序列化器
    configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    // 配置实际的反序列化器实现类
    configProps.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class.getName());
    configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, KafkaAvroDeserializer.class.getName());
    
    configProps.put(ConsumerConfig.GROUP_ID_CONFIG,"group_json");
    configProps.put("auto.offset.reset", "earliest");
    configProps.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
    configProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");
    
    return new DefaultKafkaConsumerFactory<String, BookAvro>(configProps);
}

三、额外验证点 #

1. 确认Schema Registry一致性
   确保生产者和消费者连接的是同一个Schema Registry实例，并且生产者发送消息时对应的Schema已经成功注册（可以通过访问http://localhost:8081/subjects/bookavro-value/versions查看已注册的Schema）。
2. 检查依赖版本
   确保生产者和消费者项目中spring-kafka、confluent-kafka-avro-serializer等依赖的版本保持一致，避免因版本不兼容导致的问题。

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