mongodb cdc hotfix (#17)

* [mongodb][hotfix] Fix authentication failed when using a non admin auth source and duplicated usernames apache#1934 (apache#1935) * [hotfix][mongodb] Fix resume token not found apache#1879 (apache#1938) --------- Co-authored-by: Jiabao Sun <jiabao.sun@xtransfer.cn>
BKBASE-Plugin · Mar 22, 2023 · fdea66d · fdea66d
1 parent 8649e55
commit fdea66d
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 3 deletions.
diff --git a/flink-connector-mongodb-cdc/pom.xml b/flink-connector-mongodb-cdc/pom.xml
@@ -67,7 +67,7 @@ under the License.
         <dependency>
             <groupId>org.mongodb</groupId>
             <artifactId>mongodb-driver-sync</artifactId>
-            <version>4.3.1</version>
+            <version>4.3.4</version>
         </dependency>
 
         <!-- test dependencies on Flink -->

diff --git a/...cdc/src/main/java/com/ververica/cdc/connectors/mongodb/source/dialect/MongoDBDialect.java b/...cdc/src/main/java/com/ververica/cdc/connectors/mongodb/source/dialect/MongoDBDialect.java
@@ -120,8 +120,13 @@ public static TableChange collectionSchema(TableId tableId) {
     @Override
     public ChangeStreamOffset displayCurrentOffset(MongoDBSourceConfig sourceConfig) {
         MongoClient mongoClient = clientFor(sourceConfig);
-        BsonDocument startupResumeToken =
-                getLatestResumeToken(mongoClient, ChangeStreamDescriptor.deployment());
+        CollectionDiscoveryInfo discoveryInfo = discoverAndCacheDataCollections(sourceConfig);
+        ChangeStreamDescriptor changeStreamDescriptor =
+                getChangeStreamDescriptor(
+                        sourceConfig,
+                        discoveryInfo.getDiscoveredDatabases(),
+                        discoveryInfo.getDiscoveredCollections());
+        BsonDocument startupResumeToken = getLatestResumeToken(mongoClient, changeStreamDescriptor);
 
         ChangeStreamOffset changeStreamOffset;
         if (startupResumeToken != null) {