Kafka Connect和Mongo源连接器具有管道中的查找聚合阶段

问题描述

我在mongoDB中具有以下三个集合：

//Drinkers
{
    "drinkerId" : "ID1","name" : "name","lastName" : "lastname"
}

//Drinks
{
    "drinkId" : "ID2","drinkName" : "wine","alcoholPercent" : 0.12
}
//Happy-Hours
{
    "drinkerId" : "ID1","drinkId" : "ID1"
}

我还通过以下mongo管道在欢乐时光集合上提交了一个kafka连接器实例。

    [
      {
        "$match": {
          "operationType": "insert"
        }
      },{
        "$lookup": {
          "from": "drinkers","localField": "fullDocument.drinkerId","foreignField": "drinkerId","as": "drinkerInfo"
        }
      },{
        "$unwind": "$drinkerInfo"
      },{
        "$lookup": {
          "from": "drinks","localField": "fullDocument.drinkId","foreignField": "drinkId","as": "drinkInfo"
        }
      },{
        "$unwind": "$drinkInfo"
      }
    ]

当我查看kafka-connect日志时，看到以下消息：

WARN无法恢复更改流：不允许在$ lookup中输入 $ changeStream管道20

消息未发布到该主题。

我的实际用例是，我将在饮酒者和饮料收藏者中插入较大的mongodb文档，而我想在欢乐时光收藏中插入较小的文档。最终消费者需要喝酒者和饮料信息。

我是否有另一种方法来检索饮酒者和饮料信息并获得发布给该主题的详细消息？

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）