public static ChangeStreamDescriptor getChangeStreamDescriptor(
            MongoDBSourceConfig sourceConfig,
            List<String> discoveredDatabases,
            List<String> discoveredCollections) {
        List<String> databaseList = sourceConfig.getDatabaseList();
        List<String> collectionList = sourceConfig.getCollectionList();

        ChangeStreamDescriptor changeStreamFilter;
        if (collectionList != null) {
            // Watching collections changes
            if (isIncludeListExplicitlySpecified(collectionList, discoveredCollections)) {
                changeStreamFilter =
                        ChangeStreamDescriptor.collection(
                                TableId.parse(discoveredCollections.get(0)));
            } else {
                Pattern namespaceRegex = includeListAsFlatPattern(collectionList);
                if (databaseList != null) {
                    if (isIncludeListExplicitlySpecified(databaseList, discoveredDatabases)) {
                        changeStreamFilter =
                                ChangeStreamDescriptor.database(
                                        discoveredDatabases.get(0), namespaceRegex);
                    } else {
                        Pattern databaseRegex = includeListAsFlatPattern(databaseList);
                        changeStreamFilter =
                                ChangeStreamDescriptor.deployment(databaseRegex, namespaceRegex);
                    }
                } else {
                    changeStreamFilter = ChangeStreamDescriptor.deployment(null, namespaceRegex);
                }
            }
        } else if (databaseList != null) {
            if (isIncludeListExplicitlySpecified(databaseList, discoveredDatabases)) {
                changeStreamFilter = ChangeStreamDescriptor.database(discoveredDatabases.get(0));
            } else {
                Pattern databaseRegex = includeListAsFlatPattern(databaseList);
                changeStreamFilter = ChangeStreamDescriptor.deployment(databaseRegex);
            }
        } else {
            // Watching all changes on the cluster
            changeStreamFilter = ChangeStreamDescriptor.deployment();
        }
        return changeStreamFilter;
    }
Utilities of MongoDB operations.
Copy from com.ververica:flink-connector-mongodb-cdc:2.3.0.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/utils/MongoUtils.java