使用KafkaListener监听Azure EventHub Avro事件时遇CompletionException错误

解决Azure Schema Registry Kafka Avro Deserializer升级后的NoClassDefFoundError问题 #

问题原因

升级azure-schemaregistry-kafka-avro到1.1.1版本后，该依赖对azure-core的版本要求发生变化，项目可能缺少对应版本的azure-core依赖，或存在依赖版本冲突，导致运行时找不到com.azure.core.models.MessageContent类。此外，新版本反序列化器对配置完整性要求更高，缺失必要配置也会引发间接错误。

解决方案

1. 补充兼容版本的azure-core依赖
   在pom.xml中明确引入与azure-schemaregistry-kafka-avro 1.1.1兼容的azure-core版本（推荐1.24.0，与1.1.1版本的schema registry依赖匹配）：

   <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
       <version>1.24.0</version>
   </dependency>
   

2. 完善反序列化配置项
   新版本KafkaAvroDeserializer需要完整的Schema Registry认证及配置信息，仅指定反序列化类不足以正常工作。补充核心配置示例：

   @Bean
   public ConsumerFactory<String, Object> consumerFactory() {
       Map<String, Object> props = new HashMap<>();
       // Event Hub Kafka端点
       props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-eventhub-namespace.servicebus.windows.net:9093");
       props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
       props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
       
       // Schema Registry基础配置
       props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "https://your-schema-registry-name.azurewebsites.net");
       props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true); // 使用自定义Avro类设为true，通用记录设为false
       
       // AD认证配置（使用服务主体时必填）
       props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_AZURE_CLIENT_ID_CONFIG, "your-service-principal-client-id");
       props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_AZURE_CLIENT_SECRET_CONFIG, "your-service-principal-secret");
       props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_AZURE_TENANT_ID_CONFIG, "your-azure-tenant-id");
       
       return new DefaultKafkaConsumerFactory<>(props);
   }
   

3. 排查并解决依赖冲突
   使用mvn dependency:tree命令生成依赖树，检查是否存在多个版本的azure-core。若有冲突，在azure-schemaregistry-kafka-avro依赖中排除旧版本的azure-core，再手动引入兼容版本：

   <dependency>
       <groupId>com.microsoft.azure</groupId>
       <artifactId>azure-schemaregistry-kafka-avro</artifactId>
       <version>1.1.1</version>
       <exclusions>
           <exclusion>
               <groupId>com.azure</groupId>
               <artifactId>azure-core</artifactId>
           </exclusion>
       </exclusions>
   </dependency>
   

4. 验证ByteDeserializer的正确使用
   若替换为ByteDeserializer仍报错，需确保@KafkaListener的泛型类型与反序列化器匹配。示例：

   @KafkaListener(topics = "your-eventhub-name", groupId = "consumer-group-id")
   public void listen(ConsumerRecord<String, byte[]> record) {
       // 打印byte数组长度验证消息是否正常接收
       System.out.println("Received message length: " + record.value().length);
   }
   

   若能正常接收byte数组，说明问题出在Avro反序列化环节，回到前三个步骤排查。

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