我正在尝试使用 debezium mongodb 连接器设置从 mongodb 到 kudu 的同步。但正如 debezium doc 所说，我自己也尝试过，发现 debezium mongodb CDC update/$set 消息没有过滤器（_id 值）。
{
"after": null,
"patch": "{\"$v\" : 1,\"$set\" : {\"_upts_ratio_average_points\" : {\"$numberLong\" : \"1564645156749\"},\"updatets\" : {\"$numberLong\" : \"1564645156749\"}}}",
"source": {
"version": "0.9.5.Final",
"connector": "mongodb",
"name": "promongodbdeb05",
"rs": "mgset-13056897",
"ns": "strtest.mg_jsd_result_all",
"sec": 1564645156,
"ord": 855,
"h": -1570214265415439167,
"initsync": false
},
"op": "u",
"ts_ms": 1564648181536
}
我不明白为什么要这样设计，没有过滤器真的不知道更新了哪个文档。我下载了这个连接器的源代码并尝试修复它。看起来类io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope是使用此类代码提取 MongoDB 操作日志消息的地方。而且这个文件混淆了两者_id和id操作，看起来连接器的提交者确实试图_id在 CDC 更新消息中包含值。重新构建和部署连接器后，我尝试将 CDC 消息更改valueDocument.append("id", keyDocument.get("id"));为valueDocument.append("id", keyDocument.get("_id"));仍然没有_id值。
任何熟悉 debezium 的人都可以帮助我吗？
{
private BsonDocument getUpdateDocument(R patchRecord, BsonDocument keyDocument) {
BsonDocument valueDocument = new BsonDocument();
BsonDocument document = BsonDocument.parse(patchRecord.value().toString());
if (document.containsKey("$set")) {
valueDocument = document.getDocument("$set");
}
if (document.containsKey("$unset")) {
Set<Entry<String, BsonValue>> unsetDocumentEntry = document.getDocument("$unset").entrySet();
for (Entry<String, BsonValue> valueEntry : unsetDocumentEntry) {
// In case unset of a key is false we don't have to do anything with it,
// if it's true we want to set the value to null
if (!valueEntry.getValue().asBoolean().getValue()) {
continue;
}
valueDocument.append(valueEntry.getKey(), new BsonNull());
}
}
if (!document.containsKey("$set") && !document.containsKey("$unset")) {
if (!document.containsKey("_id")) {
throw new ConnectException("Unable to process Mongo Operation, a '$set' or '$unset' is necessary " +
"for partial updates or '_id' is expected for full Document replaces.");
}
// In case of a full update we can use the whole Document as it is
// see https://docs.mongodb.com/manual/reference/method/db.collection.update/#replace-a-document-entirely
valueDocument = document;
valueDocument.remove("_id");
}
if (!valueDocument.containsKey("id")) {
valueDocument.append("id", keyDocument.get("id"));
}
if (flattenStruct) {
final BsonDocument newDocument = new BsonDocument();
valueDocument.forEach((fKey, fValue) -> newDocument.put(fKey.replace(".", delimiter), fValue));
valueDocument = newDocument;
}
return valueDocument;
}
}
@jiri，非常感谢您的回复，我收到的信息总是这样：
{ "after": null, "patch": "{\"$v\" : 1,\"$set\" : {\"_upts_ratio_average_points\" : {\"$numberLong\" : \"1564645156749\"},\"updatets\" : {\"$numberLong\" : \"1564645156749\"}}}", "source": { "version": "0.9.5.Final", "connector": "mongodb", "name": "promongodbdeb05", "rs": "mgset-13056897", "ns": "strtest.mg_jsd_result_all", "sec": 1564645156, "ord": 855, "h": -1570214265415439167, "initsync": false }, "op": "u", "ts_ms": 1564648181536 }
我搜索并发现其他人可以像这篇文章一样获得 debezium mongodb CDC： https ://rmoff.net/2018/03/27/streaming-data-from-mongodb-into-kafka-with-kafka-connect-and- debezium/ 像这样： { "after": {\"_id\" : {\"$oid\" : \"58385328e4b001431e4e497a\"}, ....
可以看到我无法获取_id，所以我无法知道在哪个文档/记录上进行了更改，但是对于上面的帖子，作者似乎可以获取_id，也可以通过检查代码来获取_id。我在上面的帖子中使用了这两个0.9.5Final和哪个。0.7.4 rev对我来说都没有运气，总是没有 _id 值。