如何在Kafka Connect JDBC中覆盖key.serializer

Question

我正在使用kafka jdbc source connector进行mysql到kafka的连接。一切正常。现在我需要传递key.serializer和value.serializer来加密数据，如macronova所示。但我没有发现输出有任何变化。

用于启动源连接器的 POST API

curl -X POST -H "Content-Type: application/json" --data '{
"name": "jdbc-source-connector-2",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"key.serializer": "org.apache.kafka.common.serialization.StringSerializer",
"value.serializer": "org.apache.kafka.common.serialization.StringSerializer",
"tasks.max": 10,
"connection.url": "jdbc:mysql://localhost:3306/connect_test?user=roo&password=roo",
"mode": "incrementing",
"table.whitelist" : "test",
"incrementing.column.name": "id",
"timestamp.column.name": "modified",
"topic.prefix": "table-",
"poll.interval.ms": 1000
}
}' http://localhost:8083/connectors

Answer 1

连接器仅通过键和值属性采用转换器，而不采用序列化程序

如果要加密整个字符串，则需要实现自己的转换器或编辑写入数据库的代码以写入Kafka，然后使用并写入数据库以及其他下游系统