MongoSourceConnector墓碑没起作用

当使用MongoSourceConnector从MongoDB中读取数据时，我们可能需要使用墓碑机制来删除已经过期的数据。 然而，有时我们可能会发现墓碑机制并没有生效，导致已过期的数据仍然存在。

解决这个问题的一个方法是在MongoDB的集合上启用TTL索引，在TTL索引上设置合适的过期时间，在数据过期后将自动启用墓碑机制删除数据。以下是一个示例TTL索引和MongoSourceConnector配置：

首先，在MongoDB集合上启用TTL索引：

db.test.ensureIndex({ "expireAt": 1 }, { expireAfterSeconds: 0 })

然后，在MongoSourceConnector配置文件中添加以下内容：

{ "name": "mongo-source-connector", "config": { "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector", "tasks.max": "1", "database": "test", "collection": "mycoll", "topic.prefix": "mytopic", "transforms": "TombstoneMark", "transforms.TombstoneMark.type": "org.apache.kafka.connect.transforms.MarkAsTombstone$Value", "transforms.TombstoneMark.fields": "expireAt" } }

注意，我们使用了“transforms”配置来启用Kafka连接器中的变换操作。在本例中，我们使用“MarkAsTombstone”变换，该变换将在满足指定条件时在消息中添加墓碑标记。我们使用“expireAt”字段来指定消息过期时间。

通过以上设置，MongoSourceConnector将能够自动使用墓碑机制删除过期的数据，并从Kafka主题中去除已过期的消息。