如何配置MongoDbConnector将多集合发布到单个Kafka主题

要让多个MongoDB集合的变更统一发送到单个Kafka主题，你需要给Debezium连接器添加主题路由转换（SMT），具体修改配置如下：

修改后的完整配置 #

{
  "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
  "collection.include.list": "dbname.messages,dbname.comments",
  "mongodb.password": "mongodbpassword",
  "tasks.max": "1",
  "database.history.kafka.topic": "dev.dbhistory.unwrap_with_key_id_8",
  "mongodb.user": "mongodbuser",
  "heartbeat.interval.ms": "90000",
  "mongodb.name": "analytics",
  "snapshot.delay.ms": "120000",
  "key.converter.schemas.enable": "false",
  "poll.interval.ms": "3000",
  "value.converter.schemas.enable": "false",
  "mongodb.authsource": "admin",
  "errors.tolerance": "all",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "mongodb.hosts": "rs0/ip:27017",
  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "database.include.list": "dbname",
  "snapshot.mode": "initial",
  // 新增主题路由配置
  "transforms": "routeToSingleTopic",
  "transforms.routeToSingleTopic.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.routeToSingleTopic.regex": "analytics.dbname.(messages|comments)",
  "transforms.routeToSingleTopic.replacement": "analytics.dbname.combined_changes"
}

关键配置说明 #

• transforms: 启用名为routeToSingleTopic的转换规则
• transforms.routeToSingleTopic.type: 指定使用Kafka Connect内置的正则路由转换类
• transforms.routeToSingleTopic.regex: 匹配原本生成的两个主题名（analytics.dbname.messages和analytics.dbname.comments）
• transforms.routeToSingleTopic.replacement: 替换成你自定义的单个目标主题名，示例中是analytics.dbname.combined_changes，可根据业务需求修改

如果后续需要将dbname下的所有集合变更都统一到单个主题，可把正则简化为analytics.dbname.*，无需逐一指定集合名。

修改配置后重启连接器，新的变更数据和初始快照数据都会同步到指定的单个Kafka主题中。

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