从 Kafka Connect 迁移 MongoDB

使用本指南将Kafka部署从社区创建的Kafka Connect MongoDB接收器Connector迁移到官方MongoDB Kafka Connector。

以下部分列出了必须对 Kafka Connect connector 配置设置和自定义类进行的更改，以转换到 MongoDB Kafka connector。

注意

Atlas Stream Processing提供MongoDB原生工具来持续进程流媒体数据、验证模式并将视图物化为Atlas数据库集合或Apache Kafka主题。

To learn more about Atlas Stream Processing, see Atlas Stream Processing.

更新配置设置

对 Kafka Connect 部署的配置设置进行以下更改，然后再将其用于 MongoDB Kafka Connector 部署：

将包含包at.grahsl.kafka.connect.mongodb的值替换为包com.mongodb.kafka.connect 。
将connector.class设置替换为 MongoDB Kafka connector 类。
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
从 Kafka Connect 属性名称中删除mongodb.前缀。例如，将mongodb.connection.uri更改为connection.uri 。
Remove the document.id.strategies setting if it exists. If the value of this setting reference custom strategies, move them to the document.id.strategy setting. Read the Update Custom Classes section to discover what changes you must make to your custom classes.
将用于指定按主题或集合覆盖的任何包含mongodb.collection前缀的属性名称替换为接收器连接器 Kafka 主题配置主题属性中的等效键。

如果在 Kafka Connect Sink Connector 部署中使用任何自定义类，请在将它们添加到 MongoDB Kafka Connector 部署之前对它们进行以下更改：

将包含at.grahsl.kafka.connect.mongodb的导入替换为com.mongodb.kafka.connect 。
将对MongoDbSinkConnector类的引用替换为MongoSinkConnector类。
更新自定义sink connector策略类以实施com.mongodb.kafka.connect.sink.processor.id.strategy.IdStrategy接口。
更新对MongoDbSinkConnectorConfig类的引用。在 MongoDB Kafka Connector 中，该类的逻辑被分割为以下类：
- MongoSinkConfig
- MongoSinkTopicConfig

如果您的类在Kafka Connect Connector部署中帖子处理器进行了子类化，更新覆盖Kafka Connect PostProcessor 类中方法的方法，以匹配MongoDB Kafka Connector PostProcessor 类的方法签名。