JSONConverter的Kafka连接模式格式

Question

我正在使用Kafka从模式注册表中检索现有模式，然后尝试使用(org.apache.kafka.connect.json.JSONConverter). JSONConverter转换返回的模式字符串

不幸的是，我从JSONConverter中得到了一个错误：

org.apache.kafka.connect.errors.DataException: Unknown schema type: object

我查看了JSONConverter代码，出现了错误，因为模式注册表返回的模式" type“是"object”(见下文)，但JSONConverter不识别该类型。

问题：

1. 检索到的模式是否可用于JSONConverter？如果是，我是否在使用这个incorrectly?
2. Is JSONConverter，希望使用不同的格式？如果是，是否有人知道JSONConverter所期望的格式？
3. 是否有不同的方法将模式注册表响应协调成一个“模式”？

以下是相关的工件：

模式注册表响应(当查询特定模式时)：

[{"subject":"test-schema","version":1,"id":1,"schemaType":"JSON","schema":"{\"title\":\"test-schema\",\"type\":\"object\",\"required\":[\"id\"],\"additionalProperties\":false,\"properties\":{\"id\":{\"type\":\"integer\"}}}"}]

当对上面的文本稍作清理时，相关的模式组件(" schema ")如下所示：

{
  "title":"test-schema",
  "type":"object",
  "required":["id"],
  "additionalProperties":false,
  "properties":{"id":{"type":"integer"}}
}

Answer 1

org.apache.kafka.connect.json.JSONConverter实际上没有使用"JSONSchema“规范。它有自己的(不是很好的文档)格式。它也根本不与模式注册表集成。

对象是struct类型。- https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/#json-schemas

如果您打算使用实际的JSONSchema (和注册表)，则需要使用来自Confluent - io.confluent.connect.json.JsonSchemaConverter的转换器。

是否有一种将模式注册表响应协调为“模式”的不同方法

如果您使用Schema客户端，那么可以使用getSchemaById方法，那么响应的schemaType()和rawSchema()方法应该可以使您接近所需的内容。这样，您将把它传递给某个JSONSchema库(例如，注册表使用的org.everit.json.schema )。