[hotfix][mongodb] Fix resume token not found apache#1879 (apache#1938)

rawlinxx · May 26, 2023 · 2be5dc4 · 2be5dc4
1 parent 1371f48
commit 2be5dc4
Showing 1 changed file with 7 additions and 2 deletions.
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) {