KafkaJDBCSinkConnector模式异常：JsonConverter使用schemas.enable需要'schema”和'payload”。

这个问题通常是由于Kafka Connect配置中缺少必要的参数而引起的。在Sink Connector中，需要配置正确的转换器和JDBC连接以正确地将数据写入目标数据库。

一种解决方法是在Kafka Connect配置文件中增加以下参数：

"transforms": "InsertTimestamp, unwrap, flatten", "transforms.InsertTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value", "transforms.InsertTimestamp.timestamp.field": "timestamp", "transforms.InsertTimestamp.timestamp.format": "yyyyMMddHHmmss", "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope", "transforms.unwrap.drop.tombstones": "false", "transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",

这里示例配置文件的完整示例：

{ "name": "jdbc-sink", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "topics": "orders", "connection.url": "jdbc:mysql://localhost:3306/mydb", "connection.user": "root", "connection.password": "root", "auto.create": "true", "auto.evolve": "true", "insert.mode": "upsert", "pk.mode": "record_value", "pk.fields": "orderId", "transforms": "InsertTimestamp, unwrap, flatten", "transforms.InsertTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value", "transforms.InsertTimestamp.timestamp.field": "timestamp", "transforms.InsertTimestamp.timestamp.format": "yyyyMMddHHmmss", "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope", "transforms.unwrap.drop.tombstones": "false", "transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter.schemas.enable": "false" } }

这里的主要改动是在value.converter参数中将schemas.enable设置为false，并添加了三个转换器函数以支持数据解包和扁平化。这些函数可以确保正确处理Kafka Connect消息中可能存在的嵌套结构。