我正在使用这个debezium-examples
source.json
{
"name": "inventory-connector",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "mysql",
"database.port": "3306",
"database.user": "debezium",
"database.password": "dbz",
"database.server.id": "184054",
"database.server.name": "dbserver1",
"database.whitelist": "inventory",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "schema-changes.inventory",
"transforms": "route",
"transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
"transforms.route.replacement": "$3"
}
}
JDBC-sink.json
{
"name": "jdbc-sink",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": "1",
"topics": "customers",
"connection.url": "jdbc:postgresql://postgres:5432/inventory?user=postgresuser&password=postgrespw",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
"auto.create": "true",
"insert.mode": "upsert",
"pk.fields": "id",
"pk.mode": "record_value"
}
}
它的工作正常。但是当我在下面的场景中进行了一些改变时。它给了我'SinkRecordField'错误。
脚本
我已从源中更改此属性
"transforms.route.replacement": "my-$2"
现在它在kafka中创建主题如下
my-inventory
当我在jdbc-sink中指定topic = my-inventory
时,它给了我以下异常[io.confluent.connect.jdbc.sink.DbStructure]
connect_1 | 2019-01-29 10:34:32,218 INFO || Unable to find fields [SinkRecordField{schema=Schema{STRING}, name='email', isPrimaryKey=false}, SinkRecordField{schema=Schema{STRING}, name='first_name', isPrimaryKey=false}, SinkRecordField{schema=Schema{STRING}, name='last_name', isPrimaryKey=false}] among column names [street, customer_id, city, state, id, type, zip] [io.confluent.connect.jdbc.sink.DbStructure]
connect_1 | 2019-01-29 10:34:32,220 ERROR || WorkerSinkTask{id=jdbc-sink-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. [org.apache.kafka.connect.runtime.WorkerSinkTask]
connect_1 | org.apache.kafka.connect.errors.ConnectException: Cannot ALTER to add missing field SinkRecordField{schema=Schema{STRING}, name='email', isPrimaryKey=false}, as it is not optional and does not have a default value
connect_1 | at io.confluent.connect.jdbc.sink.DbStructure.amendIfNecessary(DbStructure.java:133)
注意:在Db中,它创建名为“my-inventory”的表
JDBC接收器每个主题需要一个表,每个主题也有一个单独的模式(列名称为x类型)。
您在Debezium / source端的正则表达式路由实际上是将inventory
数据库中的任何表(可能包括一些系统的表,虽然我不记得是配置中的默认值)转储到“my-inventory”主题。
因此,只要您在该主题中捕获了多个表格,您就可能遇到麻烦......