[mongodb] Match multiple database and collection names using a regular expression in MongoDB. (apache#940)

Co-authored-by: Bobby Richard <[email protected]>

Author: 2 people

docs/content/connectors/mongodb-cdc.md

@@ -151,17 +151,19 @@ Connector Options
     </tr>
     <tr>
       <td>database</td>
-      <td>required</td>
+      <td>optional</td>
       <td style="word-wrap: break-word;">(none)</td>
       <td>String</td>
-      <td>Name of the database to watch for changes.</td>
+      <td>Name of the database to watch for changes. If not set then all databases will be captured. <br>
+          The database also supports regular expressions to monitor multiple databases matching the regular expression.</td>
     </tr>
     <tr>
       <td>collection</td>
-      <td>required</td>
+      <td>optional</td>
       <td style="word-wrap: break-word;">(none)</td>
       <td>String</td>
-      <td>Name of the collection in the database to watch for changes.</td>
+      <td>Name of the collection in the database to watch for changes. If not set then all collections will be captured.<br>
+          The collection also supports regular expressions to monitor multiple collections matching fully-qualified collection identifiers.</td>
     </tr>
     <tr>
       <td>connection.options</td>
@@ -324,7 +326,7 @@ This can filter only required data and improve the use of indexes by the copying
 In the following example, the `$match` aggregation operator ensures that only documents in which the closed field is set to false are copied.
 
 ```
-copy.existing.pipeline=[ { "$match": { "closed": "false" } } ]
+'copy.existing.pipeline' = '[ { "$match": { "closed": "false" } } ]'
 ```
 
 ### Change Streams

flink-cdc-e2e-tests/src/test/java/com/ververica/cdc/connectors/tests/MongoE2eITCase.java

@@ -30,6 +30,7 @@
 import com.ververica.cdc.connectors.tests.utils.FlinkContainerTestEnvironment;
 import com.ververica.cdc.connectors.tests.utils.JdbcProxy;
 import com.ververica.cdc.connectors.tests.utils.TestUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.bson.Document;
 import org.bson.types.ObjectId;
 import org.junit.After;
@@ -82,14 +83,12 @@ public void before() {
         // please check your '/etc/hosts' file contains the line 'internet_ip(not 127.0.0.1)
         // hostname' e.g: '30.225.0.87   leonard.machine'
         mongodb =
-                new MongoDBContainer()
-                        .withNetwork(NETWORK)
+                new MongoDBContainer(NETWORK)
                         .withNetworkAliases(INTER_CONTAINER_MONGO_ALIAS)
                         .withLogConsumer(new Slf4jLogConsumer(LOG));
 
         Startables.deepStart(Stream.of(mongodb)).join();
 
-        executeCommandFileInMongoDB("mongo_setup", "admin");
         MongoClientSettings settings =
                 MongoClientSettings.builder()
                         .applyConnectionString(
@@ -225,7 +224,7 @@ private String executeCommandFileInMongoDB(String fileNameIgnoreSuffix, String d
             String command1 =
                     Files.readAllLines(Paths.get(ddlTestFile.toURI())).stream()
                             .map(String::trim)
-                            .filter(x -> !x.startsWith("//") && !x.isEmpty())
+                            .filter(x -> StringUtils.isNotBlank(x) && !x.trim().startsWith("//"))
                             .map(
                                     x -> {
                                         final Matcher m = COMMENT_PATTERN.matcher(x);

flink-cdc-e2e-tests/src/test/resources/ddl/mongo_setup.js

@@ -36,9 +36,13 @@
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Locale;
 import java.util.Properties;
 
+import static com.ververica.cdc.connectors.mongodb.internal.MongoDBConnectorSourceTask.COLLECTION_INCLUDE_LIST;
+import static com.ververica.cdc.connectors.mongodb.internal.MongoDBConnectorSourceTask.DATABASE_INCLUDE_LIST;
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -123,10 +127,9 @@ public static class Builder<T> {
         private String hosts;
         private String username;
         private String password;
-        private String database;
-        private String collection;
+        private List<String> databaseList;
+        private List<String> collectionList;
         private String connectionOptions;
-        private String pipeline;
         private Integer batchSize;
         private Integer pollAwaitTimeMillis = POLL_AWAIT_TIME_MILLIS_DEFAULT;
         private Integer pollMaxBatchSize = POLL_MAX_BATCH_SIZE_DEFAULT;
@@ -167,24 +170,18 @@ public Builder<T> password(String password) {
             return this;
         }
 
-        /** Name of the database to watch for changes. */
-        public Builder<T> database(String database) {
-            this.database = database;
-            return this;
-        }
-
-        /** Name of the collection in the database to watch for changes. */
-        public Builder<T> collection(String collection) {
-            this.collection = collection;
+        /** Regular expressions list that match database names to be monitored. */
+        public Builder<T> databaseList(String... databaseList) {
+            this.databaseList = Arrays.asList(databaseList);
             return this;
         }
 
         /**
-         * An array of objects describing the pipeline operations to run. eg. [{"$match":
-         * {"operationType": "insert"}}, {"$addFields": {"Kafka": "Rules!"}}]
+         * Regular expressions that match fully-qualified collection identifiers for collections to
+         * be monitored. Each identifier is of the form {@code <databaseName>.<collectionName>}.
          */
-        public Builder<T> pipeline(String pipeline) {
-            this.pipeline = pipeline;
+        public Builder<T> collectionList(String... collectionList) {
+            this.collectionList = Arrays.asList(collectionList);
             return this;
         }
 
@@ -353,11 +350,17 @@ public DebeziumSourceFunction<T> build() {
                     "connector.class", MongoDBConnectorSourceConnector.class.getCanonicalName());
             props.setProperty("name", "mongodb_binlog_source");
 
+            ConnectionString connectionString = buildConnectionUri();
             props.setProperty(
-                    MongoSourceConfig.CONNECTION_URI_CONFIG, String.valueOf(buildConnectionUri()));
+                    MongoSourceConfig.CONNECTION_URI_CONFIG, String.valueOf(connectionString));
+
+            if (databaseList != null) {
+                props.setProperty(DATABASE_INCLUDE_LIST, String.join(",", databaseList));
+            }
 
-            props.setProperty(MongoSourceConfig.DATABASE_CONFIG, checkNotNull(database));
-            props.setProperty(MongoSourceConfig.COLLECTION_CONFIG, checkNotNull(collection));
+            if (collectionList != null) {
+                props.setProperty(COLLECTION_INCLUDE_LIST, String.join(",", collectionList));
+            }
 
             props.setProperty(MongoSourceConfig.FULL_DOCUMENT_CONFIG, FULL_DOCUMENT_UPDATE_LOOKUP);
             props.setProperty(
@@ -372,10 +375,6 @@ public DebeziumSourceFunction<T> build() {
             props.setProperty(
                     MongoSourceConfig.OUTPUT_SCHEMA_VALUE_CONFIG, OUTPUT_SCHEMA_VALUE_DEFAULT);
 
-            if (pipeline != null) {
-                props.setProperty(MongoSourceConfig.PIPELINE_CONFIG, pipeline);
-            }
-
             if (batchSize != null) {
                 props.setProperty(MongoSourceConfig.BATCH_SIZE_CONFIG, String.valueOf(batchSize));
             }

flink-connector-mongodb-cdc/src/main/java/com/ververica/cdc/connectors/mongodb/MongoDBSource.java

@@ -18,6 +18,11 @@
 
 package com.ververica.cdc.connectors.mongodb.internal;
 
+import com.mongodb.ConnectionString;
+import com.mongodb.MongoNamespace;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.kafka.connect.source.MongoSourceConfig;
 import com.mongodb.kafka.connect.source.MongoSourceTask;
 import io.debezium.connector.AbstractSourceInfo;
 import io.debezium.connector.SnapshotRecord;
@@ -29,21 +34,45 @@
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
 import org.apache.kafka.connect.source.SourceTaskContext;
+import org.bson.conversions.Bson;
 import org.bson.json.JsonReader;
 
 import java.lang.reflect.Field;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static com.mongodb.client.model.Aggregates.match;
+import static com.mongodb.client.model.Filters.and;
+import static com.mongodb.client.model.Filters.regex;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.ADD_NS_FIELD;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.ADD_NS_FIELD_NAME;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.bsonListToJson;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.collectionNames;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.collectionsFilter;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.completionPattern;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.databaseFilter;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.databaseNames;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.includeListAsPatterns;
+import static com.ververica.cdc.connectors.mongodb.utils.CollectionDiscoveryUtils.isIncludeListExplicitlySpecified;
 
 /**
  * Source Task that proxies mongodb kafka connector's {@link MongoSourceTask} to adapt to {@link
  * com.ververica.cdc.debezium.internal.DebeziumChangeFetcher}.
  */
 public class MongoDBConnectorSourceTask extends SourceTask {
 
+    public static final String DATABASE_INCLUDE_LIST = "database.include.list";
+
+    public static final String COLLECTION_INCLUDE_LIST = "collection.include.list";
+
     private static final String TRUE = "true";
 
     private static final Schema HEARTBEAT_VALUE_SCHEMA =
@@ -77,6 +106,7 @@ public void initialize(SourceTaskContext context) {
 
     @Override
     public void start(Map<String, String> props) {
+        initCapturedCollections(props);
         target.start(props);
         isInSnapshotPhase = isCopying();
     }
@@ -226,4 +256,103 @@ private boolean isCopying() {
             throw new IllegalStateException("Cannot access isCopying field of SourceTask", e);
         }
     }
+
+    private void initCapturedCollections(Map<String, String> props) {
+        ConnectionString connectionString =
+                new ConnectionString(props.get(MongoSourceConfig.CONNECTION_URI_CONFIG));
+
+        String databaseIncludeList = props.get(DATABASE_INCLUDE_LIST);
+        String collectionIncludeList = props.get(COLLECTION_INCLUDE_LIST);
+
+        List<String> databaseList =
+                Optional.ofNullable(databaseIncludeList)
+                        .map(input -> Arrays.asList(input.split(",")))
+                        .orElse(null);
+
+        List<String> collectionList =
+                Optional.ofNullable(collectionIncludeList)
+                        .map(input -> Arrays.asList(input.split(",")))
+                        .orElse(null);
+
+        if (collectionList != null) {
+            // Watching collections changes
+            List<String> discoveredDatabases;
+            List<String> discoveredCollections;
+            try (MongoClient mongoClient = MongoClients.create(connectionString)) {
+                discoveredDatabases = databaseNames(mongoClient, databaseFilter(databaseList));
+                discoveredCollections =
+                        collectionNames(
+                                mongoClient,
+                                discoveredDatabases,
+                                collectionsFilter(collectionList));
+            }
+
+            // case: database = db0, collection = coll1
+            if (isIncludeListExplicitlySpecified(collectionList, discoveredCollections)) {
+                MongoNamespace namespace = new MongoNamespace(discoveredCollections.get(0));
+                props.put(MongoSourceConfig.DATABASE_CONFIG, namespace.getDatabaseName());
+                props.put(MongoSourceConfig.COLLECTION_CONFIG, namespace.getCollectionName());
+            } else { // case: database = db0|db2, collection = (db0.coll[0-9])|(db1.coll[1-2])
+                String namespacesRegex =
+                        includeListAsPatterns(collectionList).stream()
+                                .map(Pattern::pattern)
+                                .collect(Collectors.joining("|"));
+
+                List<Bson> pipeline = new ArrayList<>();
+                pipeline.add(ADD_NS_FIELD);
+
+                Bson nsFilter = regex(ADD_NS_FIELD_NAME, namespacesRegex);
+                if (databaseList != null) {
+                    String databaseRegex =
+                            includeListAsPatterns(databaseList).stream()
+                                    .map(Pattern::pattern)
+                                    .collect(Collectors.joining("|"));
+                    Bson dbFilter = regex("ns.db", databaseRegex);
+                    nsFilter = and(dbFilter, nsFilter);
+                }
+                pipeline.add(match(nsFilter));
+
+                props.put(MongoSourceConfig.PIPELINE_CONFIG, bsonListToJson(pipeline));
+
+                String copyExistingNamespaceRegex =
+                        discoveredCollections.stream()
+                                .map(ns -> completionPattern(ns).pattern())
+                                .collect(Collectors.joining("|"));
+
+                props.put(
+                        MongoSourceConfig.COPY_EXISTING_NAMESPACE_REGEX_CONFIG,
+                        copyExistingNamespaceRegex);
+            }
+        } else if (databaseList != null) {
+            // Watching databases changes
+            List<String> discoveredDatabases;
+            try (MongoClient mongoClient = MongoClients.create(connectionString)) {
+                discoveredDatabases = databaseNames(mongoClient, databaseFilter(databaseList));
+            }
+
+            if (isIncludeListExplicitlySpecified(databaseList, discoveredDatabases)) {
+                props.put(MongoSourceConfig.DATABASE_CONFIG, discoveredDatabases.get(0));
+            } else {
+                String databaseRegex =
+                        includeListAsPatterns(databaseList).stream()
+                                .map(Pattern::pattern)
+                                .collect(Collectors.joining("|"));
+
+                List<Bson> pipeline = new ArrayList<>();
+                pipeline.add(match(regex("ns.db", databaseRegex)));
+                props.put(MongoSourceConfig.PIPELINE_CONFIG, bsonListToJson(pipeline));
+
+                String copyExistingNamespaceRegex =
+                        discoveredDatabases.stream()
+                                .map(db -> completionPattern(db + "\\..*").pattern())
+                                .collect(Collectors.joining("|"));
+
+                props.put(
+                        MongoSourceConfig.COPY_EXISTING_NAMESPACE_REGEX_CONFIG,
+                        copyExistingNamespaceRegex);
+            }
+        } else {
+            // Watching all changes on the cluster by default, we do nothing here
+        }
+    }
 }