apache-kafka-connect - 如何设置JDBC源连接器(kafka)的key？

Question

我正在从 mysql 中读取数据使用 Kafka Source JDBC 连接器 并将其发布到主题 test-mysql-petai 的数据库表。

数据库表有 2 个字段，其中 Id 是主键:

+---------+-------------+------+-----+---------+----------------+
| Field   | Type        | Null | Key | Default | Extra          |
+---------+-------------+------+-----+---------+----------------+
| id      | int(11)     | NO   | PRI | NULL    | auto_increment |
| name    | varchar(20) | YES  |     | NULL    |                |
+---------+-------------+------+-----+---------+----------------+

我需要将 id 字段的值作为主题的 Key。我尝试向 jdbc 连接器属性添加转换。

JDBCConnector.properties:

name=jdbc-source-connector    
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1   
connection.url=jdbc:mysql://127.0.0.1:3306/test?user=dins&password=pw&serverTimezone=UTC
table.whitelist=petai 
mode=incrementing
incrementing.column.name=id    
schema.pattern=""    
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=id    
topic.prefix=test-mysql-jdbc-

但是，当我使用消费者读取键和值时，我得到以下信息:

Key = {"schema":{"type":"int32","optional":false},"payload":61} 
Value ={"id":61,"name":"ttt"}

我需要得到以下信息:

Key = 61    
Value ={"id":61,"name":"ttt"}

我做错了什么？感谢您的帮助。

谢谢。

Answer 1

如果您不想在键中包含模式，您可以通过设置 key.converter.schemas.enable=false 告诉 Kafka Connect。

详细解释见Kafka Connect Deep Dive – Converters and Serialization Explained by Robin Moffatt .