如何在Debezium PostgreSQL中设置Kafka消息键为源数据库名？

当然可以把Kafka消息键设置为源数据库名，以下是具体实现方案，对应你的两个疑问：

1. 能否通过连接器配置全局设置消息键？ #

可以通过**Kafka Connect SMT（单消息转换）**在连接器层面配置消息键，无需全局集群修改。每个PostgreSQL数据库对应一个Debezium连接器实例，给每个实例添加相同的SMT配置即可实现统一的键设置逻辑。

2. 如何提取嵌套的payload.source.name作为键？ #

你需要用到两个内置的SMT组合，先从消息值中提取嵌套字段转为键，再把键简化为纯字符串：

核心配置示例 #

在每个Debezium PostgreSQL连接器的配置中添加以下内容：

# 配置SMT，将payload.source.name转为消息键
transforms=createKey,extractKey
# 第一步：从消息值中提取payload.source.name作为键的结构化数据
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.createKey.fields=payload.source.name
# 第二步：从键的结构中提取出纯字符串的数据库名
transforms.extractKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractKey.field=payload.source.name
# 指定键转换器为字符串类型，确保键是纯文本数据库名
key.converter=org.apache.kafka.connect.storage.StringConverter

配合主题路由的完整配置 #

结合你提到的“将所有事件路由到单个主题”的需求，完整的连接器配置大概是这样（每个数据库对应一个类似的配置）：

name=postgres-db1-connector
connector.class=io.debezium.connector.postgresql.PostgresConnector
# 数据库连接信息（每个连接器对应不同的数据库）
database.hostname=db1-host
database.port=5432
database.user=debezium
database.password=your-password
database.dbname=db1
database.server.name=db1-server
# 路由所有事件到同一个主题
topic.route.regex=(.*)
topic.route.replacement=all-db-changes
# SMT键配置
transforms=createKey,extractKey
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.createKey.fields=payload.source.name
transforms.extractKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractKey.field=payload.source.name
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.debezium.converters.ByteBufferConverter

额外注意事项 #

• 确保目标主题all-db-changes的分区数等于你的数据库数量，Kafka会根据消息键的哈希值将同一数据库的事件分配到同一个分区，保证事件顺序。
• payload.source.name的值对应连接器配置中的database.dbname，确认这个配置正确即可保证键是正确的数据库名。

内容的提问来源于stack exchange，提问作者mitix