ES 接收器连接器 debezium 因错误而停止

Question

要了解CDC的工作原理， 我一直在研究在debezium站点 https://debezium.io/blog/2018/03/08/creating-ddd-aggregates-with-debezium-and-kafka-streams/给出的以下eaxmple。

在这种情况下，如果我尝试将接收器连接器从 mongo db 更改为弹性搜索，然后启动 es-sink 连接器。它显示以下错误

Caused by: 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.

Mysql Debezium 源连接器属性是这个(请忽略更正 url(

{
"name": "mysql-source",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "localhost",
"database.port": "3306",
"database.user": "cdc",
"database.password": "passwrod",
"database.server.id": "1840514",
"database.server.name": "dbserver1",
"table.whitelist": "inventory.customers,inventory.addresses",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "schema-changes.inventory",
"transforms": "unwrap",
"transforms.unwrap.type":"io.debezium.transforms.UnwrapFromEnvelope",
"transforms.unwrap.drop.tombstones":"false"
}

}

此示例中使用的弹性搜索接收器连接器 https://debezium.io/blog/2018/01/17/streaming-to-elasticsearch/

弹性接收器连接器属性为以下(请忽略更正 URL(

{
"name": "elastic-sink",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "product-cdc,final_ddd_aggregates,dbserver1.inventory.customers,dbserver1.inventory.addresses",
"connection.url": "https://localhost:9243",
"transforms": "unwrap,key",
"transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
"transforms.key.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
"transforms.key.field": "id",
"key.ignore": "false",
"schema.ignore" : "true",
"value.converter.schemas.enable":"true",
"type.name":"final_ddd_aggregates"
}

}

请协助我。

Answer 1

在您的配置中，您需要按照错误消息告诉您的那样执行，并设置schemas.enable=false。使用本文中的示例，而不是：

{
"name": "mongodb-sink",
"config": {
"connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
"tasks.max": "1",
"topics": "final_ddd_aggregates",
"mongodb.connection.uri": "mongodb://mongodb:27017/inventory?w=1&journal=true",
"mongodb.collection": "customers_with_addresses",
"mongodb.document.id.strategy": "at.grahsl.kafka.connect.mongodb.processor.id.strategy.FullKeyStrategy",
"mongodb.delete.on.null.values": true
}
}

您将拥有：

{
"name": "mongodb-sink",
"config": {
"connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
"tasks.max": "1",
"topics": "final_ddd_aggregates",
"mongodb.connection.uri": "mongodb://mongodb:27017/inventory?w=1&journal=true",
"mongodb.collection": "customers_with_addresses",
"mongodb.document.id.strategy": "at.grahsl.kafka.connect.mongodb.processor.id.strategy.FullKeyStrategy",
"mongodb.delete.on.null.values": true,
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false"
}
}

要了解有关转换器等的更多信息，请参阅 https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained 和 http://rmoff.dev/ksldn19-kafka-connect

Answer 2

因为错误消息意味着您可能在正在阅读的主题中存储了没有架构的 JSON 消息。需要在源端启用它，或在接收器端禁用它。

请查看此常见问题解答条目以了解详细信息。