如何通过Kafka JDBC连接器将主键设置到消息中

Question

标题怎么说，有一种方法可以设置Kafka JDBC Source连接器从数据库读取的记录的主键？

这是来自connect-file-source.properties的配置

name=jdbc_source_postgres_foobar_01
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
connection.url=jdbc:postgresql://localhost:5432/db?user=postgres&password=root
table.whitelist=author
mode=timestamp+incrementing
incrementing.column.name=id
timestamp.column.name=entity_modify_date
validate.non.null=false
topics=migration
topic.prefix=

有没有一处房产能满足我的要求？

Answer 1

您可以通过使用Kafka Connect SMT来实现这一点，尝试添加以下配置：

transforms=createKey,extractInt
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.createKey.fields=id
transforms.extractInt.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractInt.field=specialization_id