Kafka cp-server Broker端消息验证失败原因及配置调整咨询

核心失败原因分析 #

1. JSON Schema版本不兼容 #

Confluent Schema Registry 默认仅支持JSON Schema Draft 07及更早版本，你使用的是draft/2020-12版本，这会直接导致Broker无法识别并验证Schema。

2. Schema注册或消息发送的格式问题 #

• 注册Schema时未明确指定schemaType为JSON，导致Schema Registry默认按Avro处理你的JSON Schema，后续验证自然失败。
• 发送消息时直接发送原始JSON，未使用Confluent提供的JSON Schema序列化器包装消息（缺少Schema ID等元数据），Broker无法关联到已注册的Schema进行验证。

3. Broker配置未正确生效 #

如果是在容器启动后临时修改confluent.value.schema.validation参数，而非在docker-compose.yml中预先配置，该配置不会自动生效，需要重启Broker容器才能启用验证功能。

调整配置及修复步骤 #

1. 修改JSON Schema为Draft 07版本 #

将Schema中的$schema字段替换为Draft 07的地址，其余内容可保留：

{
  "$id": "http://example.com/models/data-item-definition.json",
  "$schema": "http://json-schema.org/draft-07/schema#",
  "description": "test data item 1",
  "properties": {
    "array_val": {
      "items": {
        "type": "string"
      },
      "type": [
        "array",
        "null"
      ]
    },
    "int_val": {
      "type": "integer"
    },
    "string_val": {
      "type": "string"
    }
  },
  "required": [
    "string_val",
    "int_val"
  ],
  "title": "data item",
  "type": "object"
}

2. 正确注册JSON Schema #

使用curl注册时，明确指定schemaType为JSON，示例命令：

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{
    "schemaType": "JSON",
    "schema": "{\"$id\": \"http://example.com/models/data-item-definition.json\",\"$schema\": \"http://json-schema.org/draft-07/schema#\",\"description\": \"test data item 1\",\"properties\": {\"array_val\": {\"items\": {\"type\": \"string\"},\"type\": [\"array\",\"null\"]},\"int_val\": {\"type\": \"integer\"},\"string_val\": {\"type\": \"string\"}},\"required\": [\"string_val\",\"int_val\"],\"title\": \"data item\",\"type\": \"object\"}"
  }' \
  http://localhost:8081/subjects/your_topic-value/versions

（注意替换your_topic为你的实际主题名）

3. 使用JSON Schema序列化器发送消息 #

必须通过Confluent的JSON Schema序列化器发送消息，以确保消息包含Schema ID元数据。以Java客户端为例，配置序列化器：

value.serializer=io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer
schema.registry.url=http://localhost:8081

4. 确保Broker验证配置生效 #

在cp-all-in-one/docker-compose.yml的cp-server服务中添加环境变量，预先启用验证：

cp-server:
  image: confluentinc/cp-server:7.5.0
  environment:
    KAFKA_CONFLUENT_VALUE_SCHEMA_VALIDATION: "true"
    # 其他原有配置...

修改后重启容器：

docker-compose up -d --force-recreate cp-server

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