Micronaut集成Kafka与Avro消费者报错：Required argument未指定

从错误日志的ConsumerRecord可以看到，消息的value是JSON格式字符串而非Avro二进制数据，同时消费者报错无法绑定EasySchema类型参数，核心问题出在序列化/反序列化配置、Avro类型处理以及消费者注解缺失上，以下是具体排查和解决步骤：

1. 补全消费者类的关键注解 #

你提供的消费者代码缺少@KafkaListener注解，Micronaut无法识别该类为Kafka消费者，自然无法正确绑定消息参数。需要添加该注解：

import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.micronaut.configuration.kafka.annotation.KafkaKey;

@KafkaListener // 必须添加此注解
public class CustomerMessageConsumer {

    @Topic("with-easy-topic1")
    void receive(@KafkaKey String key, EasySchema message) {
        System.out.println(key);
        System.out.println(message);
    }
}

2. 开启消费者的Specific Avro Reader #

KafkaAvroDeserializer默认将Avro数据反序列化为GenericRecord，而非你生成的EasySchema具体类，导致Micronaut无法完成类型绑定。修改application.yml的消费者配置，添加specific.avro.reader: true：

kafka:
  consumers:
    default:
      key:
        deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      value:
        deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      schema:
        registry:
          url: http://127.0.0.1:8081
      specific:
        avro:
          reader: true # 开启该配置，反序列化直接生成具体Avro类

3. 验证生产者的Avro序列化正确性 #

错误日志显示消息value是JSON，说明生产者未正确使用Avro序列化器，需排查：

• 确认依赖完整性：确保项目引入Micronaut Kafka Avro支持和Confluent序列化器依赖（以Gradle为例）：

  implementation("io.micronaut.kafka:micronaut-kafka-avro")
  implementation("io.confluent:kafka-avro-serializer:7.4.0") // 版本需与Schema Registry匹配
  

• 检查Schema Registry状态：确认Schema Registry服务正常运行，且生产者已将EasySchema的schema注册到Registry中（可通过http://127.0.0.1:8081/subjects接口查询是否存在with-easy-topic1-value主题的schema）。
• 验证生产者配置生效：确认application.yml中生产者的序列化器配置无拼写错误，可添加日志打印生产者配置，确认KafkaAvroSerializer已被正确加载。

4. 确认Avro生成类的规范性 #

• 确保EasySchema类是通过Avro官方工具（如Maven/Gradle的Avro插件）生成的，而非手动编写。生成的类必须实现SpecificRecord接口，包含SCHEMA$静态字段和符合Avro规范的构造方法、getter/setter。
• 检查生成类的包路径与schema中的namespace（de.mydata.kafka.easy）完全一致，避免类加载或类型匹配问题。

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