Confluent Kafka结合Schema Registry的JSON消息验证配置问题

问题根因说明 #

• 开启confluent.key.schema.validation=true报错的核心原因：你当前生产者使用的Key序列化器是普通StringSerializer，不会在消息前附加Confluent Schema Registry约定的魔法字节+Schema ID前缀，Broker侧校验时无法识别该格式直接拒绝消息。
• 值侧必填字段校验不生效的原因：默认情况下KafkaJsonSchemaSerializer不会对不符合Schema的消息抛出异常，只会静默处理，必须显式开启非法Schema失败配置。

需补充的全量配置 #

情况1：不需要对Key做Schema校验（绝大多数业务场景） #

直接关闭Key的Broker侧校验即可，仅保留值侧校验：

1. 主题配置调整：将confluent.key.schema.validation设为false，仅保留confluent.value.schema.validation=true
2. 生产者补充如下配置：

// 开启非法Schema直接报错
configProps.put(KafkaJsonSchemaSerializerConfig.FAIL_INVALID_SCHEMA, "true");
// 生产环境建议关闭自动注册Schema，要求Schema提前在Registry备案
configProps.put(KafkaJsonSchemaSerializerConfig.AUTO_REGISTER_SCHEMAS, "false");
// 强制使用Schema Registry中最新版本的Schema做校验，避免本地类与远端Schema不一致
configProps.put(KafkaJsonSchemaSerializerConfig.USE_LATEST_VERSION, "true");

3. 需确认Broker服务端已经配置了schema.registry.url参数，指向你的Schema Registry地址，否则Broker无法拉取Schema做兜底校验。

情况2：需要对Key也做Schema校验 #

需完全改造Key的序列化逻辑，适配Schema校验规则：

1. 给Key定义对应的JSON Schema，在Schema Registry中注册对应{你的主题名}-key的Subject
2. 替换生产者的Key序列化器为JSON Schema序列化器：

configProps.put(
        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer");

3. 给Key对应的类也加上对应的Jackson Schema注解，同时补充和值侧一样的校验相关配置即可
4. 保持主题侧confluent.key.schema.validation=true和confluent.value.schema.validation=true配置即可

额外注意事项 #

• 如果你的业务要求禁止额外字段写入，除了Schema中配置additionalProperties=false，还可以在生产者配置中添加configProps.put(KafkaJsonSchemaSerializerConfig.FAIL_ON_UNKNOWN_PROPERTIES, "true")，提前在生产者侧拦截包含未定义字段的消息。
• 可在TestUser类上添加@JsonInclude(JsonInclude.Include.NON_NULL)注解，避免空值字段序列化后触发不必要的Schema校验。

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