如何强制Kafka消费者使用指定版本的Schema解析AVRO消息？

是可以跳过消息携带的Schema ID查询逻辑，直接使用预先指定的Schema解析所有消息，io.confluent.kafka.serializers.KafkaAvroDeserializer原生支持该能力，具体配置和注意事项如下：

实现方式 #

你可以根据自己的需求选择以下两种方案：

方案1：完全本地指定Schema，无需连接Schema Registry #

直接在消费者配置中添加avro.schema参数，赋值为你要使用的Schema的完整JSON字符串，反序列化器会优先使用该Schema解析所有消息，完全忽略消息携带的Schema ID，也不会发起任何到Schema Registry的请求。

配置示例（Java消费者）：

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的kafka地址");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "你的消费组id");
// 核心配置：指定用于反序列化的Schema
consumerProps.put("avro.schema", "{\"type\":\"record\",\"name\":\"你的记录名\",\"namespace\":\"你的命名空间\",\"fields\":[{...}]}");
// 如果你使用Avro生成的SpecificRecord实体类，可以额外开启该配置
// consumerProps.put("specific.avro.reader", "true");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");
// 部分版本的反序列化器会校验schema.registry.url是否存在，随便填占位符即可，不会实际发起请求
consumerProps.put("schema.registry.url", "http://placeholder");

方案2：自动拉取最新版本Schema，固定使用该版本解析所有消息 #

如果你不想把Schema写死在配置里，希望启动时自动从Schema Registry拉取对应Topic的最新版本Schema使用，可以配置use.latest.version=true，该模式下反序列化器只会在启动时拉取一次最新Schema缓存，后续所有消息都用该Schema解析，不会再根据消息携带的Schema ID拉取旧版本Schema。

配置示例：

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的kafka地址");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "你的消费组id");
consumerProps.put("schema.registry.url", "你的Schema Registry地址");
// 核心配置：强制使用最新版本Schema
consumerProps.put("use.latest.version", "true");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");

注意事项 #

• 无论选哪种方案，都需要保证你使用的Schema和所有历史消息的Schema兼容，符合你在Schema Registry中配置的兼容性规则（如向前兼容、全兼容），否则会出现字段类型不匹配、必填字段缺失等反序列化报错。
• avro.schema配置的优先级最高，只要配置了该参数，use.latest.version配置会自动失效。

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