Kafka-Connect JDBC Sink在更新期间报告空ID

如何解决Kafka-Connect JDBC Sink在更新期间报告空ID

我是Kafka / Kafka Connect的新手，但是在使用融合的JDBC连接器时遇到了问题。目前，我正在使用Confluent社区docker compose。

我可以成功创建一个从mysql数据库读入kafka的源代码。

curl -X POST \
-H "Content-Type: application/json" \
--data '{ "name": "college_mysql_source","config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector","tasks.max": 1,"connection.url": "jdbc:mysql://host.docker.internal:3306/...","mode": "timestamp+incrementing","timestamp.column.name": "updated_on","topic.prefix": "college_mysql_","poll.interval.ms": 1000,"table.whitelist": "college" } }' \
http://localhost:8083/connectors

数据按预期输入Kafka，Avro中正确表示了每一列。如果我通过CLI创建使用者，则可以看到数据正确。

{
    "id":112525,"pim_id":{"long":78806},"college_name":{"string":"Western University of Health Sciences"},...
}

如果我创建一个简单的JDBC接收器以将数据放入另一个mysql DB中，那么一切都很好：

curl -X POST -H "Content-Type: application/json" \
  --data '{"name": "weighted_average_mysql_sink_college","config": {"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector","tasks.max":"1","topics":"college_mysql_college","auto.create": "true","insert.mode": "insert"}}' \
  http://localhost:8083/connectors

我们正确地创建了一个表，并且所有字段（包括id）的正确填充都使新记录运行良好。但是，如果我改为创建使用插入模式upsert的接收器，则会开始出现错误。

curl -X POST -H "Content-Type: application/json" \
  --data '{"name": "weighted_average_mysql_sink_college","insert.mode": "upsert","pk.mode": "record_key","pk.fields": "id"}}' \
  http://localhost:8083/connectors

到目前为止，这确实正确创建了表，并正确地将id建立为主键，但是现在只要从主题中读取它，我们都会出错：

java.sql.BatchUpdateException: Column 'id' cannot be null

这就是我被困住的地方。主题中的数据正确地具有一个ID字段，如果我不将其声明为PK，则该ID字段将用于ID列。我尝试自己定义表，而不是让接收器创建表，弄清楚表的创建可能存在一些奇怪的问题，但是似乎无论如何我都遇到完全相同的错误。任何建议或指导都将不胜感激，我希望解决方案很简单，而我只是想为那些有更多经验的人提供一些显而易见的建议。

谢谢！

解决方法

您需要设置“ pk.mode”：“ record_value”