具有 JSON 架构的 Kafka jdbc 接收器连接器不起作用

Question

使用最新的 kafka 和 confluent jdbc 接收器连接器。发送一个非常简单的 Json 消息：

{
    "schema": {
        "type": "struct",
        "fields": [
            {
                "type": "int",
                "optional": false,
                "field": "id"
            },
            {
                "type": "string",
                "optional": true,
                "field": "msg"
            }
        ],
        "optional": false,
        "name": "msgschema"
    },
    "payload": {
        "id": 222,
        "msg": "hi"
    }
}

但是出现错误:

org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires "schema" and "payload" fields and may not contain additional fields. If you are trying to deserialize plain JSON data, set schemas.enable=false in your converter configuration.

Jsonlint说Json是有效的。我在 kafka 配置中保留了 json schemas.enable=true。有什么指示吗？

Answer 1

我最近遇到了同样的问题，我花了多次重试才弄清楚遗漏了什么：

以下设置对我有用：

key.converter.schemas.enable=false
value.converter.schemas.enable=true

此外，请确保该表以前存在于数据库中，并且连接器不应尝试创建表。设置auto.create=false

Answer 2

为了使用JDBC接收器，您的流消息必须有一个模式。这可以通过使用Avro和Schema Registry来实现，也可以通过使用JSON和Schema来实现。如果在最初运行源属性文件后配置了< code>schemas.enable=true，您可能需要删除主题，重新运行接收器，然后再次启动源端。

例：

接收器属性文件

name=sink-mysql
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=test-mysql-jdbc-foobar
connection.url=jdbc:mysql://127.0.0.1:3306/demo?user=user1&password=user1pass
auto.create=true

以及示例工作配置文件< code > connect-avro-standalone . properties :

bootstrap.servers=localhost:9092

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

# Local storage file for offset data
offset.storage.file.filename=/tmp/connect.offsets

plugin.path=share/java

并执行

./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/sink.properties

Answer 3

您需要告诉 Connect 您的架构已嵌入到您正在使用的 JSON 中。

您有：

value.converter=org.apache.kafka.connect.json.JsonConverter 

但还需要：

value.converter.schemas.enable=true