使用kafka-json-schema-console-producer同时指定键值Schema的问题咨询

问题1解答 #

是的，新增键Schema只需添加key.schema配置项即可，它的语法规则和value.schema完全一致，均遵循标准JSON Schema规范。
额外需要补充两个配套配置：

• 新增--property parse.key=true参数，告知生产者你的输入内容包含键、值两部分，需要拆分解析
• 显式指定键的序列化器为JSON Schema专用序列化器，确保键的Schema校验和注册逻辑正常生效

问题2解答 #

完整命令示例如下，你可以根据实际业务调整Schema定义、Schema注册中心地址、分隔符等配置：

kafka-json-schema-console-producer \
  --broker-list localhost:9092 \
  --topic some-topic \
  # 开启键解析，默认键值分隔符为制表符\t
  --property parse.key=true \
  # 可选配置：自定义键值分隔符
  # --property key.separator="|" \
  # 键序列化器与Schema注册中心配置
  --property key.converter=io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer \
  --property key.converter.schema.registry.url=http://localhost:8081 \
  # 键Schema定义，可根据业务调整
  --property key.schema='
{
  "type": "object",
  "required": ["snack_id"],
  "additionalProperties": false,
  "properties": {
    "snack_id": {"type": "string"}
  }
}' \
  # 值序列化器与Schema注册中心配置
  --property value.converter=io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer \
  --property value.converter.schema.registry.url=http://localhost:8081 \
  # 值Schema定义
  --property value.schema='
{
  "definitions" : {
    "record:myrecord" : {
      "type" : "object",
      "required" : [ "name", "calories" ],
      "additionalProperties" : false,
      "properties" : {
        "name" : {"type" : "string"},
        "calories" : {"type" : "number"},
        "colour" : {"type" : "string"}
      }
    }
  },
  "$ref" : "#/definitions/record:myrecord"
}' < snacks_with_key.txt

配套的输入文件snacks_with_key.txt每行格式参考如下（键JSON与值JSON用制表符分隔）：

{"snack_id":"s001"}	{"name":"potato chips","calories":520,"colour":"yellow"}
{"snack_id":"s002"}	{"name":"chocolate","calories":580,"colour":"brown"}

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